topos: propagate via branch from receive to send event

- identify properly the branch

modules/topos/topos_mod.c

@@ -222,11 +222,7 @@ int tps_prepare_msg(sip_msg_t *msg)
 		return 2;
 	}
 
-	/* force 2nd via parsing here - it helps checking it later */
-	if (parse_headers(msg, HDR_VIA2_F, 0)==-1
-			|| (msg->via2==0) || (msg->via2->error!=PARSE_OK)) {
-		LM_DBG("no second via in this message \n");
-	}
+	parse_headers(msg, HDR_VIA2_F, 0);
 
 	if(parse_headers(msg, HDR_CSEQ_F, 0)!=0 || msg->cseq==NULL) {
 		LM_ERR("cannot parse cseq header\n");
@@ -345,6 +341,11 @@ int tps_msg_sent(void *data)
 
 		tps_request_sent(&msg, dialog, local);
 	} else {
+		/* reply */
+		if(msg.first_line.u.reply.statuscode==100) {
+			/* nothing to do - it should be locally generated */
+			return 0;
+		}
 		tps_response_sent(&msg);
 	}
 

modules/topos/tps_msg.c

@@ -47,6 +47,8 @@
 
 extern int _tps_param_mask_callid;
 
+str _sr_hname_xbranch = str_init("P-SR-XBranch");
+
 /**
  *
  */
@@ -256,7 +258,7 @@ int tps_skip_msg(sip_msg_t *msg)
 /**
  *
  */
-int tps_pack_request(sip_msg_t *msg, tps_data_t *ptsd)
+int tps_pack_message(sip_msg_t *msg, tps_data_t *ptsd)
 {
 	hdr_field_t *hdr;
 	via_body_t *via;
@@ -268,6 +270,7 @@ int tps_pack_request(sip_msg_t *msg, tps_data_t *ptsd)
 	if(ptsd->cp==NULL) {
 		ptsd->cp = ptsd->cbuf;
 	}
+
 	i = 0;
 	for(hdr=msg->h_via1; hdr; hdr=next_sibling_hdr(hdr)) {
 		for(via=(struct via_body*)hdr->parsed; via; via=via->next) {
@@ -385,9 +388,11 @@ int tps_pack_request(sip_msg_t *msg, tps_data_t *ptsd)
 		ptsd->s_rr.len = ptsd->a_rr.len;
 		ptsd->a_rr.len = 0;
 	}
-	LM_DBG("compacted headers - a_rr: [%.*s](%d) - b_rr: [%.*s](%d)\n",
+	LM_DBG("compacted headers - a_rr: [%.*s](%d) - b_rr: [%.*s](%d)"
+			" - s_rr: [%.*s](%d)\n",
 			ptsd->a_rr.len, ZSW(ptsd->a_rr.s), ptsd->a_rr.len,
-			ptsd->b_rr.len, ZSW(ptsd->b_rr.s), ptsd->b_rr.len);
+			ptsd->b_rr.len, ZSW(ptsd->b_rr.s), ptsd->b_rr.len,
+			ptsd->s_rr.len, ZSW(ptsd->s_rr.s), ptsd->s_rr.len);
 	LM_DBG("compacted headers - as_contact: [%.*s](%d) - bs_contact: [%.*s](%d)\n",
 			ptsd->as_contact.len, ZSW(ptsd->as_contact.s), ptsd->as_contact.len,
 			ptsd->bs_contact.len, ZSW(ptsd->bs_contact.s), ptsd->bs_contact.len);
@@ -425,6 +430,31 @@ int tps_reinsert_contact(sip_msg_t *msg, tps_data_t *ptsd, str *hbody)
 	return 0;
 }
 
+/**
+ *
+ */
+int tps_remove_name_headers(sip_msg_t *msg, str *hname)
+{
+	hdr_field_t *hf;
+	struct lump* l;
+	for (hf=msg->headers; hf; hf=hf->next)
+	{
+		if (hf->name.len==hname->len
+				&& strncasecmp(hf->name.s, hname->s,
+					hname->len)==0)
+		{
+			l=del_lump(msg, hf->name.s-msg->buf, hf->len, 0);
+			if (l==0) {
+				LM_ERR("unable to delete header [%.*s]\n",
+						hname->len, hname->s);
+				return -1;
+			}
+			return 0;
+		}
+	}
+	return 0;
+}
+
 /**
  *
  */
@@ -439,6 +469,57 @@ int tps_reappend_via(sip_msg_t *msg, tps_data_t *ptsd, str *hbody)
 	return 0;
 }
 
+/**
+ *
+ */
+int tps_append_xbranch(sip_msg_t *msg, str *hbody)
+{
+	if(tps_add_headers(msg, &_sr_hname_xbranch, hbody, 0)<0) {
+		LM_ERR("failed to add xbranch header [%.*s]/%d\n",
+				hbody->len, hbody->s, hbody->len);
+		return -1;
+	}
+
+	return 0;
+}
+
+/**
+ *
+ */
+int tps_remove_xbranch(sip_msg_t *msg)
+{
+	return tps_remove_name_headers(msg, &_sr_hname_xbranch);
+}
+
+/**
+ *
+ */
+int tps_get_xbranch(sip_msg_t *msg, str *hbody)
+{
+	hdr_field_t *hf;
+	if(parse_headers(msg, HDR_EOH_F, 0)<0) {
+		return -1;
+	}
+	if(tps_add_headers(msg, &_sr_hname_xbranch, hbody, 0)<0) {
+		return -1;
+	}
+
+	for (hf=msg->headers; hf; hf=hf->next)
+	{
+		if(_sr_hname_xbranch.len==hf->name.len
+				&& strncasecmp(_sr_hname_xbranch.s, hf->name.s,
+					hf->name.len)==0) {
+			break;
+		}
+	}
+	if(hf!=NULL) {
+		*hbody = hf->body;
+		return 0;
+	}
+	return -1;
+}
+
+
 /**
  *
  */
@@ -474,6 +555,8 @@ int tps_response_received(sip_msg_t *msg)
 	tps_data_t stsd;
 	tps_data_t btsd;
 	str lkey;
+	str ftag;
+	uint32_t direction;
 
 	if(msg->first_line.u.reply.statuscode==100) {
 		/* nothing to do - it should be absorbed */
@@ -486,7 +569,7 @@ int tps_response_received(sip_msg_t *msg)
 
 	lkey = msg->callid->body;
 
-	if(tps_pack_request(msg, &mtsd)<0) {
+	if(tps_pack_message(msg, &mtsd)<0) {
 		LM_ERR("failed to extract and pack the headers\n");
 		return -1;
 	}
@@ -494,15 +577,33 @@ int tps_response_received(sip_msg_t *msg)
 	if(tps_storage_load_branch(msg, &mtsd, &btsd)<0) {
 		goto error;
 	}
-	if(tps_storage_update_branch(msg, &mtsd, &btsd)<0) {
-		goto error;
-	}
 	LM_DBG("loaded dialog a_uuid [%.*s]\n",
 			btsd.a_uuid.len, ZSW(btsd.a_uuid.s));
 	if(tps_storage_load_dialog(msg, &btsd, &stsd)<0) {
 		goto error;
 	}
-	if(tps_storage_update_dialog(msg, &btsd, &stsd)<0) {
+
+	/* detect direction - get from-tag */
+	if(parse_from_header(msg)<0 || msg->from==NULL) {
+		LM_ERR("failed getting 'from' header!\n");
+		goto error;
+	}
+	ftag = get_from(msg)->tag_value;
+
+	if(stsd.a_tag.len!=ftag.len) {
+		direction = TPS_DIR_UPSTREAM;
+	} else {
+		if(memcpy(stsd.a_tag.s, ftag.s, ftag.len)==0) {
+			direction = TPS_DIR_DOWNSTREAM;
+		} else {
+			direction = TPS_DIR_UPSTREAM;
+		}
+	}
+	mtsd.direction = direction;
+	if(tps_storage_update_branch(msg, &mtsd, &btsd)<0) {
+		goto error;
+	}
+	if(tps_storage_update_dialog(msg, &mtsd, &stsd)<0) {
 		goto error;
 	}
 	tps_storage_lock_release(&lkey);
@@ -510,6 +611,7 @@ int tps_response_received(sip_msg_t *msg)
 	tps_reappend_via(msg, &btsd, &btsd.x_via);
 	tps_reappend_rr(msg, &btsd, &btsd.s_rr);
 	tps_reappend_rr(msg, &btsd, &btsd.x_rr);
+	tps_append_xbranch(msg, &mtsd.x_vbranch1);
 
 	return 0;
 
@@ -533,7 +635,7 @@ int tps_request_sent(sip_msg_t *msg, int dialog, int local)
 	memset(&stsd, 0, sizeof(tps_data_t));
 	ptsd = &mtsd;
 
-	if(tps_pack_request(msg, &mtsd)<0) {
+	if(tps_pack_message(msg, &mtsd)<0) {
 		LM_ERR("failed to extract and pack the headers\n");
 		return -1;
 	}
@@ -591,15 +693,23 @@ int tps_response_sent(sip_msg_t *msg)
 	tps_data_t btsd;
 	str lkey;
 	int direction = TPS_DIR_UPSTREAM;
+	str xvbranch;
 
 	memset(&mtsd, 0, sizeof(tps_data_t));
 	memset(&stsd, 0, sizeof(tps_data_t));
 	memset(&btsd, 0, sizeof(tps_data_t));
 
-	if(tps_pack_request(msg, &mtsd)<0) {
+	if(tps_get_xbranch(msg, &xvbranch)<0) {
+		LM_DBG("no x-branch header - nothing to do\n");
+		return 0;
+	}
+
+	if(tps_pack_message(msg, &mtsd)<0) {
 		LM_ERR("failed to extract and pack the headers\n");
 		return -1;
 	}
+	mtsd.x_vbranch1 = xvbranch;
+	tps_remove_xbranch(msg);
 
 	if(get_cseq(msg)->method_id==METHOD_MESSAGE) {
 		tps_remove_headers(msg, HDR_RECORDROUTE_T);

modules/topos/tps_storage.c

@@ -268,22 +268,24 @@ int tps_storage_link_msg(sip_msg_t *msg, tps_data_t *td, int dir)
 	trim(&stxt);
 	td->a_callid = stxt;
 
+	/* get from-tag */
+	if(parse_from_header(msg)<0 || msg->from==NULL) {
+		LM_ERR("failed getting 'from' header!\n");
+		goto error;
+	}
+	td->a_tag = get_from(msg)->tag_value;
+
+	/* get to-tag */
+	if(parse_to_header(msg)<0 || msg->to==NULL) {
+		LM_ERR("failed getting 'to' header!\n");
+		goto error;
+	}
+	td->b_tag = get_to(msg)->tag_value;
+
 	if(dir==TPS_DIR_DOWNSTREAM) {
-		/* get from-tag */
-		if(parse_from_header(msg)<0 || msg->from==NULL) {
-			LM_ERR("failed getting 'from' header!\n");
-			goto error;
-		}
-		td->a_tag = get_from(msg)->tag_value;
 		td->x_tag = td->a_tag;
 	} else {
-		/* get to-tag */
-		if(parse_to_header(msg)<0 || msg->to==NULL) {
-			LM_ERR("failed getting 'to' header!\n");
-			goto error;
-		}
-		td->b_tag = get_to(msg)->tag_value;
-		td->b_tag = td->a_tag;
+		td->x_tag = td->b_tag;
 	}
 
 	td->x_via = td->x_via2;
@@ -370,8 +372,8 @@ str td_col_iflags = str_init("iflags");
 str td_col_a_uri = str_init("a_uri");
 str td_col_b_uri = str_init("b_uri");
 str td_col_r_uri = str_init("r_uri");
-str td_col_a_srcip = str_init("a_srcip");
-str td_col_b_srcip = str_init("b_srcip");
+str td_col_a_srcaddr = str_init("a_srcaddr");
+str td_col_b_srcaddr = str_init("b_srcaddr");
 str td_col_s_method = str_init("s_method");
 str td_col_s_cseq = str_init("s_cseq");
 
@@ -494,14 +496,14 @@ int tps_db_insert_dialog(tps_data_t *td)
 	db_vals[nr_keys].val.str_val = TPS_STRZ(td->r_uri);
 	nr_keys++;
 
-	db_keys[nr_keys] = &td_col_a_srcip;
+	db_keys[nr_keys] = &td_col_a_srcaddr;
 	db_vals[nr_keys].type = DB1_STR;
-	db_vals[nr_keys].val.str_val = TPS_STRZ(td->a_srcip);
+	db_vals[nr_keys].val.str_val = TPS_STRZ(td->a_srcaddr);
 	nr_keys++;
 
-	db_keys[nr_keys] = &td_col_b_srcip;
+	db_keys[nr_keys] = &td_col_b_srcaddr;
 	db_vals[nr_keys].type = DB1_STR;
-	db_vals[nr_keys].val.str_val = TPS_STRZ(td->b_srcip);
+	db_vals[nr_keys].val.str_val = TPS_STRZ(td->b_srcaddr);
 	nr_keys++;
 
 	db_keys[nr_keys] = &td_col_s_method;
@@ -850,7 +852,15 @@ int tps_storage_load_dialog(sip_msg_t *msg, tps_data_t *md, tps_data_t *sd)
 	db_ops[nr_keys]=OP_EQ;
 	db_vals[nr_keys].type = DB1_STR;
 	db_vals[nr_keys].nul = 0;
-	db_vals[nr_keys].val.str_val = TPS_STRZ(md->a_uuid);
+	if(md->a_uuid.len>0 && md->a_uuid.s[0]=='a') {
+		db_vals[nr_keys].val.str_val = TPS_STRZ(md->a_uuid);
+	} else {
+		if(md->b_uuid.len<=0) {
+			LM_ERR("no valid dlg uuid\n");
+			return -1;
+		}
+		db_vals[nr_keys].val.str_val = TPS_STRZ(md->b_uuid);
+	}
 	nr_keys++;
 
 	db_cols[nr_cols++] = &td_col_rectime;
@@ -870,8 +880,8 @@ int tps_storage_load_dialog(sip_msg_t *msg, tps_data_t *md, tps_data_t *sd)
 	db_cols[nr_cols++] = &td_col_a_uri;
 	db_cols[nr_cols++] = &td_col_b_uri;
 	db_cols[nr_cols++] = &td_col_r_uri;
-	db_cols[nr_cols++] = &td_col_a_srcip;
-	db_cols[nr_cols++] = &td_col_b_srcip;
+	db_cols[nr_cols++] = &td_col_a_srcaddr;
+	db_cols[nr_cols++] = &td_col_b_srcaddr;
 	db_cols[nr_cols++] = &td_col_s_method;
 	db_cols[nr_cols++] = &td_col_s_cseq;
 
@@ -913,8 +923,8 @@ int tps_storage_load_dialog(sip_msg_t *msg, tps_data_t *md, tps_data_t *sd)
 	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->a_uri); n++;
 	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->b_uri); n++;
 	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->r_uri); n++;
-	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->a_srcip); n++;
-	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->b_srcip); n++;
+	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->a_srcaddr); n++;
+	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->b_srcaddr); n++;
 	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->s_method); n++;
 	TPS_DATA_APPEND_DB(sd, db_res, n, &sd->s_cseq); n++;
 
@@ -943,6 +953,73 @@ int tps_storage_update_branch(sip_msg_t *msg, tps_data_t *md, tps_data_t *sd)
  */
 int tps_storage_update_dialog(sip_msg_t *msg, tps_data_t *md, tps_data_t *sd)
 {
+	db_key_t db_keys[4];
+	db_op_t  db_ops[4];
+	db_val_t db_vals[4];
+	db_key_t db_ucols[TPS_NR_KEYS];
+	db_val_t db_uvals[TPS_NR_KEYS];
+	db1_res_t* db_res = NULL;
+	int nr_keys;
+	int nr_ucols;
+	int n;
+	int ret;
+
+	if(msg==NULL || md==NULL || sd==NULL || _tps_db_handle==NULL)
+		return -1;
+
+	if(md->s_method_id != METHOD_INVITE) {
+		return 0;
+	}
+	if(msg->first_line.u.reply.statuscode<200
+			|| msg->first_line.u.reply.statuscode>=300) {
+		return 0;
+	}
+
+	ret = tps_storage_link_msg(msg, md, md->direction);
+	if(ret<0) return -1;
+
+	memset(db_ucols, 0, TPS_NR_KEYS*sizeof(db_key_t));
+	memset(db_uvals, 0, TPS_NR_KEYS*sizeof(db_val_t));
+
+	nr_keys = 0;
+	nr_ucols = 0;
+
+	db_keys[nr_keys]=&td_col_a_uuid;
+	db_ops[nr_keys]=OP_EQ;
+	db_vals[nr_keys].type = DB1_STR;
+	db_vals[nr_keys].nul = 0;
+	if(sd->a_uuid.len>0 && sd->a_uuid.s[0]=='a') {
+		db_vals[nr_keys].val.str_val = TPS_STRZ(sd->a_uuid);
+	} else {
+		if(sd->b_uuid.len<=0) {
+			LM_ERR("no valid dlg uuid\n");
+			return -1;
+		}
+		db_vals[nr_keys].val.str_val = TPS_STRZ(sd->b_uuid);
+	}
+	nr_keys++;
+
+	db_ucols[nr_ucols] = &td_col_b_contact;
+	db_uvals[nr_ucols].type = DB1_STR;
+	db_uvals[nr_ucols].val.str_val = TPS_STRZ(md->b_contact);
+	nr_ucols++;
+
+	db_ucols[nr_ucols] = &td_col_b_rr;
+	db_uvals[nr_ucols].type = DB1_STR;
+	db_uvals[nr_ucols].val.str_val = TPS_STRZ(md->b_rr);
+	nr_ucols++;
+
+	if (_tpsdbf.use_table(_tps_db_handle, &td_table_name) < 0) {
+		LM_ERR("failed to perform use table\n");
+		return -1;
+	}
+
+	if(_tpsdbf.update(_tps_db_handle, db_keys, db_ops, db_vals,
+				db_ucols, db_uvals, nr_keys, nr_ucols)!=0) {
+		LM_ERR("failed to do db update for [%.*s]!\n",
+				md->a_uuid.len, md->a_uuid.s);
+		return -1;
+	}
 	return 0;
 }
 

modules/topos/tps_storage.h

@@ -53,8 +53,10 @@ typedef struct tps_data {
 	str a_uri;
 	str b_uri;
 	str r_uri;
-	str a_srcip;
-	str b_srcip;
+	str a_srcaddr;
+	str b_srcaddr;
+	str a_socket;
+	str b_socket;
 	str x_via1;
 	str x_via2;
 	str x_vbranch1;