msilo.c

性能优秀的SIP Proxy

	}

	if(parse_uri(pto->uri.s, pto->uri.len, &puri)!=0)
	{
		LOG(L_ERR, "MSILO:m_dump: bad R-URI!\n");
		goto error;
	}
	if(puri.user.len<=0 || puri.user.s==NULL
			|| puri.host.len<=0 || puri.host.s==NULL)
	{
		LOG(L_ERR, "MSILO:m_dump: bad URI in To header!\n");
		goto error;
	}

	db_vals[0].type = DB_STR;
	db_vals[0].nul = 0;
	db_vals[0].val.str_val.s = puri.user.s;
	db_vals[0].val.str_val.len = puri.user.len;

	db_vals[1].type = DB_STR;
	db_vals[1].nul = 0;
	db_vals[1].val.str_val.s = puri.host.s;
	db_vals[1].val.str_val.len = puri.host.len;

	db_vals[2].type = DB_INT;
	db_vals[2].nul = 0;
	db_vals[2].val.int_val = 0;
	
	if (msilo_dbf.use_table(db_con, ms_db_table) < 0)
	{
		LOG(L_ERR, "MSILO:m_dump: Error in use_table\n");
		goto error;
	}

	if((msilo_dbf.query(db_con,db_keys,db_ops,db_vals,db_cols,db_no_keys,
				db_no_cols, NULL,&db_res)!=0) || (RES_ROW_N(db_res) <= 0))
	{
		DBG("MSILO:m_dump: no stored message for <%.*s>!\n", pto->uri.len,
					pto->uri.s);
		goto done;
	}
		
	DBG("MSILO:m_dump: dumping [%d] messages for <%.*s>!!!\n", 
			RES_ROW_N(db_res), pto->uri.len, pto->uri.s);

	for(i = 0; i < RES_ROW_N(db_res); i++) 
	{
		mid =  RES_ROWS(db_res)[i].values[0].val.int_val;
		if(msg_list_check_msg(ml, mid))
		{
			DBG("MSILO:m_dump: message[%d] mid=%d already sent.\n", 
				i, mid);
			continue;
		}
		
		memset(str_vals, 0, 4*sizeof(str));
		SET_STR_VAL(str_vals[0], db_res, i, 1); /* from */
		SET_STR_VAL(str_vals[1], db_res, i, 2); /* to */
		SET_STR_VAL(str_vals[2], db_res, i, 3); /* body */
		SET_STR_VAL(str_vals[3], db_res, i, 4); /* ctype */

		hdr_str.len = 1024;
		if(m_build_headers(&hdr_str, str_vals[3] /*ctype*/,
				str_vals[0]/*from*/) < 0)
		{
			LOG(L_ERR, "MSILO:m_dump: headers building failed [%d]\n", mid);
			if (msilo_dbf.free_result(db_con, db_res) < 0)
				DBG("MSILO:m_dump: Error while freeing result of"
					" query\n");
			msg_list_set_flag(ml, mid, MS_MSG_ERRO);
			goto error;
		}
			
		DBG("MSILO:m_dump: msg [%d-%d] for: %.*s\n", i+1, mid,
				pto->uri.len, pto->uri.s);
			
		/** sending using TM function: t_uac */
		body_str.len = 1024;
		rtime = 
			(time_t)RES_ROWS(db_res)[i].values[5/*inc time*/].val.int_val;
		n = m_build_body(&body_str, rtime, str_vals[2/*body*/], 0);
		if(n<0)
			DBG("MSILO:m_dump: sending simple body\n");
		else
			DBG("MSILO:m_dump: sending composed body\n");
		
			tmb.t_request(&msg_type,  /* Type of the message */
					&pto->uri,        /* Request-URI */
					&str_vals[1],     /* To */
					&str_vals[0],     /* From */
					&hdr_str,         /* Optional headers including CRLF */
					(n<0)?&str_vals[2]:&body_str, /* Message body */
					m_tm_callback,    /* Callback function */
					(void*)(long)mid        /* Callback parameter */
				);
	}

done:
	/**
	 * Free the result because we don't need it
	 * anymore
	 */
	if (db_res!=NULL && msilo_dbf.free_result(db_con, db_res) < 0)
		DBG("MSILO:m_dump: Error while freeing result of query\n");

	return 1;
error:
	return -1;
}

/**
 * - cleaning up the messages that got reply
 * - delete expired messages from database
 */
void m_clean_silo(unsigned int ticks, void *param)
{
	msg_list_el mle = NULL, p;
	db_key_t db_keys[MAX_DEL_KEYS];
	db_val_t db_vals[MAX_DEL_KEYS];
	db_op_t  db_ops[1] = { OP_LEQ };
	int n;
	
	DBG("MSILO:clean_silo: cleaning stored messages - %d\n", ticks);
	
	msg_list_check(ml);
	mle = p = msg_list_reset(ml);
	n = 0;
	while(p)
	{
		if(p->flag & MS_MSG_DONE)
		{
#ifdef STATISTICS
			if(p->flag & MS_MSG_TSND)
				update_stat(ms_dumped_msgs, 1);
			else
				update_stat(ms_dumped_rmds, 1);
#endif

			db_keys[n] = sc_mid;
			db_vals[n].type = DB_INT;
			db_vals[n].nul = 0;
			db_vals[n].val.int_val = p->msgid;
			DBG("MSILO:clean_silo: cleaning sent message [%d]\n", p->msgid);
			n++;
			if(n==MAX_DEL_KEYS)
			{
				if (msilo_dbf.delete(db_con, db_keys, NULL, db_vals, n) < 0) 
					DBG("MSILO:clean_silo: error cleaning %d messages.\n",n);
				n = 0;
			}
		}
		if((p->flag & MS_MSG_ERRO) && (p->flag & MS_MSG_TSND))
		{ /* set snd time to 0 */
			ms_reset_stime(p->msgid);
#ifdef STATISTICS
			update_stat(ms_failed_rmds, 1);
#endif

		}
#ifdef STATISTICS
		if((p->flag & MS_MSG_ERRO) && !(p->flag & MS_MSG_TSND))
			update_stat(ms_failed_msgs, 1);
#endif
		p = p->next;
	}
	if(n>0)
	{
		if (msilo_dbf.delete(db_con, db_keys, NULL, db_vals, n) < 0) 
			DBG("MSILO:clean_silo: error cleaning %d messages\n", n);
		n = 0;
	}

	msg_list_el_free_all(mle);
	
	/* cleaning expired messages */
	if(ticks%(ms_check_time*ms_clean_period)<ms_check_time)
	{
		DBG("MSILO:clean_silo: cleaning expired messages\n");
		db_keys[0] = sc_exp_time;
		db_vals[0].type = DB_INT;
		db_vals[0].nul = 0;
		db_vals[0].val.int_val = (int)time(NULL);
		if (msilo_dbf.delete(db_con, db_keys, db_ops, db_vals, 1) < 0) 
			DBG("MSILO:clean_silo: ERROR cleaning expired messages\n");
	}
}


/**
 * destroy function
 */
void destroy(void)
{
	DBG("MSILO: destroy module ...\n");
	msg_list_free(ml);

	if(db_con && msilo_dbf.close)
		msilo_dbf.close(db_con);
}

/** 
 * TM callback function - delete message from database if was sent OK
 */
void m_tm_callback( struct cell *t, int type, struct tmcb_params *ps)
{
	if(ps->param==NULL || *ps->param==0)
	{
		DBG("MSILO m_tm_callback: message id not received\n");
		goto done;
	}
	
	LOG(L_DBG, "MSILO:m_tm_callback: completed with status %d [mid: %ld/%d]\n",
		ps->code, (long)ps->param, *((int*)ps->param));
	if(!db_con)
	{
		LOG(L_ERR, "MSILO:m_tm_callback: db_con is NULL\n");
		goto done;
	}
	if(ps->code >= 300)
	{
		LOG(L_DBG,
			"MSILO:m_tm_callback: message <%d> was not sent successfully\n",
			*((int*)ps->param));
		msg_list_set_flag(ml, *((int*)ps->param), MS_MSG_ERRO);
		goto done;
	}

	LOG(L_DBG, "MSILO:m_tm_callback: message <%d> was sent successfully\n",
		*((int*)ps->param));
	msg_list_set_flag(ml, *((int*)ps->param), MS_MSG_DONE);

done:
	return;
}

void m_send_ontimer(unsigned int ticks, void *param)
{
	db_key_t db_keys[2];
	db_op_t  db_ops[2];
	db_val_t db_vals[2];
	db_key_t db_cols[6];
	db_res_t* db_res = NULL;
	int i, db_no_cols = 6, db_no_keys = 2, mid, n;
	static char hdr_buf[1024];
	static char uri_buf[1024];
	static char body_buf[1024];
	str puri;

	str str_vals[4], hdr_str , body_str;
	time_t stime;

	if(ms_reminder.s==NULL)
	{
		LOG(L_WARN,"MSILO:m_send_ontimer: weird - reminder address null\n");
		return;
	}
	
	/* init */
	db_keys[0]=sc_snd_time;
	db_keys[1]=sc_snd_time;
	db_ops[0]=OP_NEQ;
	db_ops[1]=OP_LEQ;

	db_cols[0]=sc_mid;
	db_cols[1]=sc_uri_user;
	db_cols[2]=sc_uri_host;
	db_cols[3]=sc_body;
	db_cols[4]=sc_ctype;
	db_cols[5]=sc_snd_time;

	
	DBG("MSILO:m_send_ontimer: ------------ start ------------\n");
	hdr_str.s=hdr_buf;
	hdr_str.len=1024;
	body_str.s=body_buf;
	body_str.len=1024;
	
	db_vals[0].type = DB_INT;
	db_vals[0].nul = 0;
	db_vals[0].val.int_val = 0;
	
	db_vals[1].type = DB_INT;
	db_vals[1].nul = 0;
	db_vals[1].val.int_val = (int)time(NULL);
	
	if (msilo_dbf.use_table(db_con, ms_db_table) < 0)
	{
		LOG(L_ERR, "MSILO:m_send_ontimer: Error in use_table\n");
		return;
	}

	if((msilo_dbf.query(db_con,db_keys,db_ops,db_vals,db_cols,db_no_keys,
				db_no_cols, NULL,&db_res)!=0) || (RES_ROW_N(db_res) <= 0))
	{
		DBG("MSILO:m_send_ontimer: no message for <%.*s>!\n",
				24, ctime((time_t*)(&db_vals[1].val.int_val)));
		goto done;
	}
		
	DBG("MSILO:m_send_ontimer: dumping [%d] messages for <%.*s>!!!\n", 
			RES_ROW_N(db_res), 24, ctime((time_t*)(&db_vals[1].val.int_val)));

	for(i = 0; i < RES_ROW_N(db_res); i++) 
	{
		mid =  RES_ROWS(db_res)[i].values[0].val.int_val;
		if(msg_list_check_msg(ml, mid))
		{
			DBG("MSILO:m_send_ontimer: message[%d] mid=%d already sent.\n", 
				i, mid);
			continue;
		}
		
		memset(str_vals, 0, 4*sizeof(str));
		SET_STR_VAL(str_vals[0], db_res, i, 1); /* user */
		SET_STR_VAL(str_vals[1], db_res, i, 2); /* host */
		SET_STR_VAL(str_vals[2], db_res, i, 3); /* body */
		SET_STR_VAL(str_vals[3], db_res, i, 4); /* ctype */

		hdr_str.len = 1024;
		if(m_build_headers(&hdr_str, str_vals[3] /*ctype*/,
				ms_reminder/*from*/) < 0)
		{
			LOG(L_ERR, "MSILO:m_send_ontimer: headers building failed [%d]\n",
					mid);
			if (msilo_dbf.free_result(db_con, db_res) < 0)
				DBG("MSILO:m_send_ontimer: Error while freeing result of"
					" query\n");
			msg_list_set_flag(ml, mid, MS_MSG_ERRO);
			return;
		}

		puri.s = uri_buf;
		puri.len = 4 + str_vals[0].len + 1 + str_vals[1].len;
		memcpy(puri.s, "sip:", 4);
		memcpy(puri.s+4, str_vals[0].s, str_vals[0].len);
		puri.s[4+str_vals[0].len] = '@';
		memcpy(puri.s+4+str_vals[0].len+1, str_vals[1].s, str_vals[1].len);
		
		DBG("MSILO:m_send_ontimer: msg [%d-%d] for: %.*s\n", i+1, mid,
				puri.len, puri.s);
			
		/** sending using TM function: t_uac */
		body_str.len = 1024;
		stime = 
			(time_t)RES_ROWS(db_res)[i].values[5/*snd time*/].val.int_val;
		n = m_build_body(&body_str, 0, str_vals[2/*body*/], stime);
		if(n<0)
			DBG("MSILO:m_send_ontimer: sending simple body\n");
		else
			DBG("MSILO:m_send_ontimer: sending composed body\n");
		
		msg_list_set_flag(ml, mid, MS_MSG_TSND);
		
		tmb.t_request(&msg_type,  /* Type of the message */
					&puri,            /* Request-URI */
					&puri,            /* To */
					&ms_reminder,     /* From */
					&hdr_str,         /* Optional headers including CRLF */
					(n<0)?&str_vals[2]:&body_str, /* Message body */
					m_tm_callback,    /* Callback function */
					(void*)(long)mid        /* Callback parameter */
				);
	}

done:
	/**
	 * Free the result because we don't need it anymore
	 */
	if (db_res!=NULL && msilo_dbf.free_result(db_con, db_res) < 0)
		DBG("MSILO:m_send_ontimer: Error while freeing result of query\n");

	return;
}

int ms_reset_stime(int mid)
{
	db_key_t db_keys[1];
	db_op_t  db_ops[1];
	db_val_t db_vals[1];
	db_key_t db_cols[1];
	db_val_t db_cvals[1];
	
	db_keys[0]=sc_mid;
	db_ops[0]=OP_EQ;

	db_vals[0].type = DB_INT;
	db_vals[0].nul = 0;
	db_vals[0].val.int_val = mid;
	

	db_cols[0]=sc_snd_time;
	db_cvals[0].type = DB_INT;
	db_cvals[0].nul = 0;
	db_cvals[0].val.int_val = 0;
	
	DBG("MSILO:ms_reset_stime: updating send time for [%d]!\n", mid);
	
	if (msilo_dbf.use_table(db_con, ms_db_table) < 0)
	{
		LOG(L_ERR, "MSILO:ms_reset_stime: Error in use_table\n");
		return -1;
	}

	if(msilo_dbf.update(db_con,db_keys,db_ops,db_vals,db_cols,db_cvals,1,1)!=0)
	{
		LOG(L_ERR, "MSILO:ms_reset_stime: error making update for [%d]!\n",
				mid);
		return -1;
	}
	return 0;
}