gfwleak/av-frag-rssb
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

2018.10.22 v4.0//alter g_sip_opt_type and add enum sip_opt_index

Browse Source 
2018.11.02	   v4.0//add timer_on
2018.11.05	   v4.0//alter voip save_media frg_len>headerlen
2018.11.07    v4.0//alter json_log opt name for k
This commit is contained in:
dumeijied
2018-11-07 14:39:11 +08:00
parent d95b034457
commit 256c0223b6
9 changed files with 201 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/frag_dedup.c
View File

@@ -486,6 +486,7 @@ void proc_media_multisrc(media_t* mdi, int timeout)
context,
soqav_query_free_timeout,
&mdi->timer_idx);
g_frag_run.multisrc_timer_on[mdi->thread_seq] = 1;
if(NULL!=mdi->opt[MEDIA_OPT_URL][mdi->url_opt_index])
{
query_detail->url = (char*)malloc(mdi->opt[MEDIA_OPT_URL][mdi->url_opt_index]->opt_len);

19
src/frag_reassembly.c
View File

@@ -410,6 +410,7 @@ void set_sip_query_task(frag_unit_t* frg_unit, media_t* mdi)
context,
index_query_timeout_free,
&mdi->index_query_timer_idx);
g_frag_run.index_query_timer_on[mdi->thread_seq] = 1;
}
}
}
@@ -467,6 +468,7 @@ void save_media_opt(frag_unit_t* frg_unit, media_t* mdi)
context,
index_query_timeout_free,
&mdi->index_query_timer_idx);
g_frag_run.index_query_timer_on[mdi->thread_seq] = 1;
}
}
}
@@ -677,12 +679,12 @@ int media_create(frag_unit_t* frg_unit)
long rec_cb = 0;
/*<2A><><EFBFBD>鶨ʱ<E9B6A8><CAB1>*/
if(g_frag_cfg.av_dedup_switch && g_frag_run.multisrc_timer[frg_unit->thread_seq])
if(g_frag_cfg.av_dedup_switch && g_frag_run.multisrc_timer_on[frg_unit->thread_seq] && g_frag_run.multisrc_timer[frg_unit->thread_seq])
{
MESA_timer_check(g_frag_run.multisrc_timer[frg_unit->thread_seq], time(NULL), g_frag_cfg.multisrc_timer_cb_maxtime);
}
/*<2A><><EFBFBD>鶨ʱ<E9B6A8><CAB1>*/
if(g_frag_run.index_query_timer[frg_unit->thread_seq])
if(g_frag_run.index_query_timer_on[frg_unit->thread_seq] && g_frag_run.index_query_timer[frg_unit->thread_seq])
{
MESA_timer_check(g_frag_run.index_query_timer[frg_unit->thread_seq], time(NULL), g_frag_cfg.index_query_timer_cb_maxtime);
}
@@ -1060,11 +1062,14 @@ long media_preproc_cb(void *data, const uint8_t *key, uint size, void *user_arg)
{
if(g_frag_cfg.cpz_type == CPZ_VOIP)
{
uint32_t datalen = frg->datalen - sizeof(voip_header_t);
char* data = (char*)calloc(1,datalen);
memcpy(data,frg->data + sizeof(voip_header_t),datalen);
IVI_seg_t* a_ivi_seg = IVI_seg_malloc(frg->offset, frg->offset + datalen - 1, data);
IVI_insert(mdi->save_ivi,a_ivi_seg);
if(frg->datalen > sizeof(voip_header_t))
{
uint32_t datalen = frg->datalen - sizeof(voip_header_t);
char* data = (char*)calloc(1,datalen);
memcpy(data,frg->data + sizeof(voip_header_t),datalen);
IVI_seg_t* a_ivi_seg = IVI_seg_malloc(frg->offset, frg->offset + datalen - 1, data);
IVI_insert(mdi->save_ivi,a_ivi_seg);
}
}
else
{

52
src/frag_reassembly_in.h
View File

@@ -16,7 +16,7 @@
#include "main.h"
#include "sifter.h"
#include "frag_reassembly.h"
#include "frag_reassembly_in.h"
//#include "frag_voip.h"
#define LAY_ADDR_CNT 8
#define PID_MAX_LEN 64
@@ -109,9 +109,55 @@ typedef enum
#define WEBMAIL_CONT_INFO_NUM 10
/*SIPѡ<50><D1A1><EFBFBD><EFBFBD><EFBFBD><EFBFBD>*/
#define SIP_OPT_NUM 32
typedef enum
{
SIP_HMGET_INDEX,//NOT SAVE OPT
SIP_KEY_INDEX,//NOT SAVE OPT
SIP_DATA_FLAG_INDEX,//NOT SAVE OPT
SIP_RTP_4TUPLE_OPT_INDEX,//NOT SAVE OPT
SIP_URI_OPT_INDEX,
SIP_FROM_OPT_INDEX,
SIP_TO_OPT_INDEX,
SIP_SGATEWAY_OPT_INDEX,//NOT NEED SENDBACK
SIP_CGATEWAY_OPT_INDEX,//NOT NEED SENDBACK
SIP_DURATION_OPT_INDEX,
SIP_S_CODING_OPT_INDEX,
SIP_C_CODING_OPT_INDEX,
SIP_FROM_TAGS_OPT_INDEX,
SIP_TO_TAGS_OPT_INDEX,
SIP_CALL_ID_OPT_INDEX,
SIP_CSEQ_OPT_INDEX,
SIP_C_CONTACT_OPT_INDEX,
SIP_S_CONTACT_OPT_INDEX,
SIP_USERAGENT_OPT_INDEX,
SIP_SERVER_OPT_INDEX,//NOT NEED SENDBACK
SIP_C_CRYPTO_OPT_INDEX,//NOT NEED SENDBACK
SIP_C_INLINE_OPT_INDEX,//NOT NEED SENDBACK
SIP_S_CRYPTO_OPT_INDEX,//NOT NEED SENDBACK
SIP_S_INLINE_OPT_INDEX,//NOT NEED SENDBACK
SIP_RINGING_OPT_INDEX,//NOT NEED SENDBACK
SIP_REASON_OPT_INDEX,
SIP_SIP_4TUPLE_OPT_INDEX,
SIP_S_VIA_OPT_INDEX,
SIP_S_RECORD_ROUTES_OPT_INDEX,
SIP_S_ROUTE_OPT_INDEX,
SIP_C_VIA_OPT_INDEX,
SIP_C_RECORD_ROUTES_OPT_INDEX,
SIP_C_ROUTE_OPT_INDEX,
SIP_RESCODE_OPT_INDEX,
SIP_CAPIP_OPT_INDEX,
SIP_OPT_NUM,
}sip_opt_index;
//#define SIP_OPT_NUM 33
/*SIP<49><50>ѯredis<69><73><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>*/
#define SIP_REDIS_CMMD_NUM 35
//#define SIP_REDIS_CMMD_NUM 35
/*SIP<49><50>ѡ<EFBFBD><D1A1>*/
typedef struct sip_opt_s
{

4
src/frag_recv.c
View File

@@ -183,6 +183,7 @@ void* bizman_recv_data_to_queue(void *param)
size = bizman_recv(g_frag_run.recv_bizman[tid],(char*)buf,sizeof(buf),&src_ip,&src_port,&stream_id,&is_complete);
if(is_complete&BIZMAN_READ_CHUNK && size>0)
{
if(g_frag_stat.sysinfo_stat[BIZMAN_RECV_QUEUE][QUEUE_CURRENT]<g_frag_cfg.bizman_queue_maxnum)
{
recv_data = (bizman_recv_data_t*)calloc(1, sizeof(bizman_recv_data_t));
@@ -190,12 +191,10 @@ void* bizman_recv_data_to_queue(void *param)
recv_data->src_ip = src_ip;
recv_data->data = (char*)malloc(size);
memcpy(recv_data->data, buf, size);
/*data[15] = the last byte of PID*/
tid = recv_data->data[15]%g_frag_cfg.thread_num;
lq_rec = MESA_lqueue_join_tail(g_frag_run.recv_bizman_lq[tid], &recv_data, sizeof(recv_data));
if(lq_rec==MESA_QUEUE_RET_OK)
{
/*add to queue stat */
atomic_inc(&g_frag_stat.sysinfo_stat[BIZMAN_RECV_QUEUE][QUEUE_IN]);
}
else
@@ -209,6 +208,7 @@ void* bizman_recv_data_to_queue(void *param)
{
atomic_inc(&g_frag_stat.stat_info[RECV_DROP][TOTAL_PKTS]);
atomic_add(&g_frag_stat.stat_info[RECV_DROP][TOTAL_BYTES], size);
}
atomic_inc(&g_frag_stat.stat_info[RECV][TOTAL_PKTS]);
atomic_add(&g_frag_stat.stat_info[RECV][TOTAL_BYTES], size);

110
src/frag_voip.c
View File

@@ -49,41 +49,45 @@ extern frag_reassembly_t frag_rssb;
int g_voip_fulllog_sock_fd = -1;
int g_voip_surveylog_sock_fd = -1;
const char* g_sip_cmmd_argv[SIP_REDIS_CMMD_NUM] =
const char* g_sip_cmmd_argv[SIP_OPT_NUM] =
{
"HMGET",
"",
"data_flag",
"RTP_4tuple",
"URI",
"From",
"From",
"To",
"S_gateway",
"C_gateway",
"Duration",
"S_coding",
"S_coding",
"C_coding",
"From_tag",
"To_tag",
"Call_ID",
"CSeq",
"CSeq",
"C_Contact",
"S_Contact",
"User_Agent",
"Server",
"C_crypto",
"C_crypto",
"C_inline",
"S_crypto",
"S_inline",
"Ringing_time",
"Reason",
"Reason",
"SIP_4tuple",
"S_Via",
"S_Record_Route",
"S_Route",
"C_Via",
"C_Via",
"C_Record_Route",
"C_Route",
"Rescode",
@@ -93,40 +97,46 @@ const char* g_sip_cmmd_argv[SIP_REDIS_CMMD_NUM] =
/*opt_type is fulllog opt_type*/
sip_opt_t g_sip_opt_type[SIP_OPT_NUM] =
{
{ "RTP_4tuple", 0xFF},
{ "HMGET", 0xFF},
{ "", 0xFF},
{ "data_flag", 0xFF},
{ "RTP_4tuple", 0xFF},
{ "URI", 0x20},
{ "From", 0x25},
{ "To", 0x26},
{ "S_gateway", 0xFF},
{ "C_gateway", 0xFF},
{ "Duration", 0x2A}, //6
{ "S_coding", 0x24}, //7
{ "C_coding", 0x23}, //8
{ "From_tag", 0xFF}, //9
{ "To_tag", 0xFF}, //10
{ "Call_ID", 0xFF}, //11
{ "CSeq", 0xFF}, //12
{ "C_Contact", 0x21}, //13
{ "S_Contact", 0x21}, //14
{ "User_Agent", 0xFF}, //15
{ "S_gateway", 0xFF},
{ "C_gateway", 0xFF},
{ "Duration", 0x2A},
{ "S_coding", 0x24},
{ "C_coding", 0x23},
{ "From_tag", 0xFF},
{ "To_tag", 0xFF},
{ "Call_ID", 0xFF},
{ "CSeq", 0xFF},
{ "C_Contact", 0x21},
{ "S_Contact", 0x21},//14
{ "User_Agent", 0xFF}, //15
{ "Server", 0xFF},
{ "C_crypto", 0xFF},
{ "C_inline", 0xFF},
{ "S_crypto", 0xFF},
{ "S_inline", 0xFF},
{ "Ringing_time", 0xFF},
{ "S_inline", 0xFF},
{ "Ringing_time", 0xFF},
{ "Reason", 0xFF}, //22
{ "SIP_4tuple", 0xFF}, //23
{ "SIP_4tuple", 0xFF}, //23
{ "S_Via", 0x22}, //24
{ "S_Record_Route", 0x27}, //25
{ "S_Record_Route", 0x27}, //25
{ "S_Route", 0x28}, //26
{ "C_Via", 0x22}, //27
{ "C_Record_Route", 0x27}, //28
{ "C_Record_Route", 0x27}, //28
{ "C_Route", 0x28}, //29
{ "Rescode", 0xFF}, //30
{ "CAPIP", 0x2C}, //31
};
@@ -499,6 +509,7 @@ int sip_send_survey_log(media_t* mdi, char* survey, uint32_t survey_len)
char sip_capip[128]={0};
if(NULL!=mdi->sip_opt[SIP_CAPIP_OPT_INDEX])
{
strcat(cap_ip_buf,",");
memcpy(sip_capip,mdi->sip_opt[SIP_CAPIP_OPT_INDEX]->opt_value,mdi->sip_opt[SIP_CAPIP_OPT_INDEX]->opt_len);
strcat(cap_ip_buf,sip_capip);
}
@@ -843,6 +854,7 @@ int sip_send_full_log(media_t* mdi)
char sip_capip[128]={0};
if(NULL!=mdi->sip_opt[SIP_CAPIP_OPT_INDEX])
{
strcat(cap_ip_buf,",");
memcpy(sip_capip,mdi->sip_opt[SIP_CAPIP_OPT_INDEX]->opt_value,mdi->sip_opt[SIP_CAPIP_OPT_INDEX]->opt_len);
strcat(cap_ip_buf,sip_capip);
}
@@ -925,7 +937,7 @@ void send_voip_full_json_log(media_t* mdi)
memset(pbuf, 0, sizeof(pbuf));
snprintf(pbuf, sizeof(pbuf), "%" PRIu64 "", mdi->lastpkt_time - mdi->create_time);
cJSON_AddStringToObject(root, "duation", pbuf);
cJSON_AddStringToObject(root, "duration", pbuf);
char src_ip[32] = {0};
char src_port[8] = {0};
@@ -987,43 +999,43 @@ void send_voip_full_json_log(media_t* mdi)
if(NULL!=mdi->sip_opt[SIP_C_CONTACT_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Contacts", mdi->sip_opt[SIP_C_CONTACT_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "contacts", mdi->sip_opt[SIP_C_CONTACT_OPT_INDEX]->opt_value);
}
else if(NULL!=mdi->sip_opt[SIP_S_CONTACT_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Contacts", mdi->sip_opt[SIP_S_CONTACT_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "contacts", mdi->sip_opt[SIP_S_CONTACT_OPT_INDEX]->opt_value);
}
if(NULL!=mdi->sip_opt[SIP_C_VIA_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Via", mdi->sip_opt[SIP_C_VIA_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "via", mdi->sip_opt[SIP_C_VIA_OPT_INDEX]->opt_value);
}
else if(NULL!=mdi->sip_opt[SIP_S_VIA_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Via", mdi->sip_opt[SIP_S_VIA_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "via", mdi->sip_opt[SIP_S_VIA_OPT_INDEX]->opt_value);
}
if(NULL!=mdi->sip_opt[SIP_C_ROUTE_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Route", mdi->sip_opt[SIP_C_ROUTE_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "route", mdi->sip_opt[SIP_C_ROUTE_OPT_INDEX]->opt_value);
}
else if(NULL!=mdi->sip_opt[SIP_S_ROUTE_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Route", mdi->sip_opt[SIP_S_ROUTE_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "route", mdi->sip_opt[SIP_S_ROUTE_OPT_INDEX]->opt_value);
}
if(NULL!=mdi->sip_opt[SIP_C_RECORD_ROUTES_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Record_route", mdi->sip_opt[SIP_C_RECORD_ROUTES_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "record_route", mdi->sip_opt[SIP_C_RECORD_ROUTES_OPT_INDEX]->opt_value);
}
else if(NULL!=mdi->sip_opt[SIP_S_RECORD_ROUTES_OPT_INDEX])
{
cJSON_AddStringToObject(root, "Record_route", mdi->sip_opt[SIP_S_RECORD_ROUTES_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "record_route", mdi->sip_opt[SIP_S_RECORD_ROUTES_OPT_INDEX]->opt_value);
}
if((NULL != mdi->sip_opt[SIP_USERAGENT_OPT_INDEX])&&(mdi->sip_opt[SIP_USERAGENT_OPT_INDEX]->opt_len > 0)&&(NULL != mdi->sip_opt[SIP_USERAGENT_OPT_INDEX]->opt_value))
{
cJSON_AddStringToObject(root, "User_agent", mdi->sip_opt[SIP_USERAGENT_OPT_INDEX]->opt_value);
cJSON_AddStringToObject(root, "user_agent", mdi->sip_opt[SIP_USERAGENT_OPT_INDEX]->opt_value);
}
char* survey = NULL;
@@ -1118,17 +1130,17 @@ int redis_sip_index_3_query(char* index_key, frag_unit_t* frg_unit, int thread_
char command[REDIS_CMMD_MAXLEN] = {0};
char* p_cmmd = command;
int cmmd_len = 0;
size_t cmmd_argvlen[SIP_REDIS_CMMD_NUM] = {0};
size_t cmmd_argvlen[SIP_OPT_NUM] = {0};
redisReply* reply = NULL;
g_sip_cmmd_argv[1] = index_key;
for(int i=0;i<SIP_REDIS_CMMD_NUM;i++)
for(int i=0;i<SIP_OPT_NUM;i++)
{
cmmd_argvlen[i] = strlen(g_sip_cmmd_argv[i]);
}
/*just for record log*/
for(int i=0;i<SIP_REDIS_CMMD_NUM;i++)
for(int i=0;i<SIP_OPT_NUM;i++)
{
snprintf(p_cmmd, REDIS_CMMD_MAXLEN-cmmd_len, "%s ", g_sip_cmmd_argv[i]);
cmmd_len = strlen(g_sip_cmmd_argv[i])+1;
@@ -1137,7 +1149,7 @@ int redis_sip_index_3_query(char* index_key, frag_unit_t* frg_unit, int thread_
MESA_handle_runtime_log(frag_rssb.logger, RLOG_LV_INFO, FRAG_REASSEMBLY_MODULE_NAME,
"{%s:%d} [before %d redis cluster exec command %" PRIu64 "='%s' succ].",
__FILE__,__LINE__, thread_seq, frg_unit->pid,command);
rec = redis_excute_command(thread_seq, frag_rssb.logger, &reply, command ,SIP_REDIS_CMMD_NUM, g_sip_cmmd_argv, cmmd_argvlen);
rec = redis_excute_command(thread_seq, frag_rssb.logger, &reply, command ,SIP_OPT_NUM, g_sip_cmmd_argv, cmmd_argvlen);
MESA_handle_runtime_log(frag_rssb.logger, RLOG_LV_INFO, FRAG_REASSEMBLY_MODULE_NAME,
"{%s:%d} [after %d redis cluster exec command %" PRIu64 "='%s' succ].",
__FILE__,__LINE__, thread_seq, frg_unit->pid,command);
@@ -1168,11 +1180,11 @@ int redis_sip_index_3_query(char* index_key, frag_unit_t* frg_unit, int thread_
}
else
{
frg_unit->sip_opt[i] = (opt_in_t*)malloc(sizeof(opt_in_t));
frg_unit->sip_opt[i]->opt_len = reply->element[i]->len;
frg_unit->sip_opt[i]->opt_value = (char*)malloc(reply->element[i]->len);
frg_unit->sip_opt[i]->opt_type = g_sip_opt_type[i].opt_type;
memcpy(frg_unit->sip_opt[i]->opt_value, reply->element[i]->str, reply->element[i]->len);
frg_unit->sip_opt[i+2] = (opt_in_t*)malloc(sizeof(opt_in_t));
frg_unit->sip_opt[i+2]->opt_len = reply->element[i]->len;
frg_unit->sip_opt[i+2]->opt_value = (char*)malloc(reply->element[i]->len);
frg_unit->sip_opt[i+2]->opt_type = g_sip_opt_type[i+2].opt_type;
memcpy(frg_unit->sip_opt[i+2]->opt_value, reply->element[i]->str, reply->element[i]->len);
}
}
}

66
src/frag_voip.h
View File

@@ -4,6 +4,54 @@
#include "frag_reassembly_in.h"
/*index in g_sip_opt_type*/
/*typedef enum
{
SIP_HMGET_INDEX,//NOT SAVE OPT
SIP_KEY_INDEX,//NOT SAVE OPT
SIP_DATA_FLAG_INDEX,//NOT SAVE OPT
SIP_RTP_4TUPLE_OPT_INDEX,//NOT SAVE OPT
SIP_URI_OPT_INDEX,
SIP_FROM_OPT_INDEX,
SIP_TO_OPT_INDEX,
SIP_SGATEWAY_OPT_INDEX,//NOT NEED SENDBACK
SIP_CGATEWAY_OPT_INDEX,//NOT NEED SENDBACK
SIP_DURATION_OPT_INDEX,
SIP_S_CODING_OPT_INDEX,
SIP_C_CODING_OPT_INDEX,
SIP_FROM_TAGS_OPT_INDEX,
SIP_TO_TAGS_OPT_INDEX,
SIP_CALL_ID_OPT_INDEX,
SIP_CSEQ_OPT_INDEX,
SIP_C_CONTACT_OPT_INDEX,
SIP_S_CONTACT_OPT_INDEX,
SIP_USERAGENT_OPT_INDEX,
SIP_SERVER_OPT_INDEX,//NOT NEED SENDBACK
SIP_C_CRYPTO_OPT_INDEX,//NOT NEED SENDBACK
SIP_C_INLINE_OPT_INDEX,//NOT NEED SENDBACK
SIP_S_CRYPTO_OPT_INDEX,//NOT NEED SENDBACK
SIP_S_INLINE_OPT_INDEX,//NOT NEED SENDBACK
SIP_RINGING_OPT_INDEX,//NOT NEED SENDBACK
SIP_REASON_OPT_INDEX,
SIP_SIP_4TUPLE_OPT_INDEX,
SIP_S_VIA_OPT_INDEX,
SIP_S_RECORD_ROUTES_OPT_INDEX,
SIP_S_ROUTE_OPT_INDEX,
SIP_C_VIA_OPT_INDEX,
SIP_C_RECORD_ROUTES_OPT_INDEX,
SIP_C_ROUTE_OPT_INDEX,
SIP_RESCODE_OPT_INDEX,
SIP_CAPIP_OPT_INDEX,
SIP_OPT_NUM,
}sip_opt_index;
#define SIP_OPT_MAXNUM 16
#define SIP_CALL_ID_OPT_INDEX 11
#define SIP_FROM_TAGS_OPT_INDEX 9
#define SIP_TO_TAGS_OPT_INDEX 10
@@ -28,6 +76,7 @@
#define SIP_SIP_4TUPLE_OPT_INDEX 23
#define SIP_DURATION_OPT_INDEX 6
#define SIP_CAPIP_OPT_INDEX 31
*/
#define SIP_PROTO_SIP "SIP-RTP"
#define SIP_PROTO_RTP "RTP"
@@ -35,7 +84,6 @@
#define SIP_PROTO_OPT "SIP"
/*ȫ<><C8AB><EFBFBD><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ѡ<EFBFBD><D1A1><EFBFBD><EFBFBD><EFBFBD><EFBFBD>*/
#define SIP_OPT_MAXNUM 16
/*voip<69><70>־<EFBFBD><D6BE>Ϣͷsip_log_msg_header_t*/
#define SIP_LOG_CONT_CODE_NOTHING 0
#define SIP_LOG_VERSION 3
@@ -47,25 +95,25 @@
#define SIP_SURVEYLOG_MSG_TYPE 0x21
/*ȫ<><C8AB><EFBFBD><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD>g_sip_opt_type*/
#define OPT_VOIP_PID 0x29
#define OPT_VOIP_PID 0x29
#define OPT_VOIP_DUATION_FULLLOG 0x2A
#define OPT_VOIP_VOICE_DIR_FULLLOG 0x2B
#define OPT_VOIP_CAP_IP_FULLLOG 0x2C
/*<2A><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>־ѡ<D6BE><D1A1><EFBFBD><EFBFBD><EFBFBD><EFBFBD>*/
#define OPT_LAYER_ADDR_V4 0x3B //0x3B:RTP, 0x2D:SIP, change
#define OPT_VOIP_RECV_TIME 0x2F
#define OPT_VOIP_DUATION 0x30
#define OPT_VOIP_PROTOCOL 0x31
#define OPT_LAYER_ADDR_V4 0x3B //0x3B:RTP, 0x2D:SIP, change
#define OPT_VOIP_RECV_TIME 0x2F
#define OPT_VOIP_DUATION 0x30
#define OPT_VOIP_PROTOCOL 0x31
#define OPT_VOIP_CALLING_ACCOUNT 0x32
#define OPT_VOIP_CALLED_ACCOUNT 0x33
#define OPT_VOIP_RELATION_RTP_LAYER_ADDR_V4 0x2D //0x3B:RTP, 0x2D:SIP, change
#define OPT_VOIP_RELATION_RTP_LAYER_ADDR_V4 0x2D //0x3B:RTP, 0x2D:SIP, change
#define OPT_VOIP_FROM_TO_STORE_IP 0x36
#define OPT_VOIP_FROM_TO_STORE_URL 0x37
#define OPT_VOIP_TO_FROM_STORE_IP 0x38
#define OPT_VOIP_TO_FROM_STORE_URL 0x39
#define OPT_VOIP_VOICE_DIR 0x3C
#define OPT_VOIP_CAP_IP 0x3D
#define OPT_VOIP_VOICE_DIR 0x3C
#define OPT_VOIP_CAP_IP 0x3D
/*sip_sendlog_flag*/
#define SIP_SEND_FULL_LOG 0x01

22
src/main.c
View File

@@ -42,9 +42,9 @@ const char* frag_rssb_version = "2018-08-13T09:00:00";
const char* frag_rssb_version_time = "2018-08-13T09:00:00";
const char* frag_rssb_version_des = "MESA@iie rssb_maskey";
int FRAG_RSSB_VERSION_1_0_20181008 = 0;
const char* frag_rssb_version_time_in = "2018-10-08";
const char* frag_rssb_version_des_in = "hard balance";
int FRAG_RSSB_VERSION_1_0_20181107 = 0;
const char* frag_rssb_version_time_in = "2018-11-07";
const char* frag_rssb_version_des_in = "alter json_log opt name for k";
void frag_rssb_history()
{
//2015.11.15 v1.0 create the project
@@ -219,6 +219,10 @@ void frag_rssb_history()
//2018.09.20 v4.0//1 voip_fulllog add voice_dir opt
//2018.09.27 v4.0 //1. hard balance
//2018.10.08 v4.0//1.add send_voip_full_json_log for K_PROJECT;2.voip_fulllog and voip_surveylog add voice_dir and cap_ip opt
//2018.10.22 v4.0//alter g_sip_opt_type and add enum sip_opt_index
//2018.11.02 v4.0//add timer_on
//2018.11.05 v4.0//alter voip save_media frg_len>headerlen
//2018.11.07 v4.0//alter json_log opt name for k
}
frag_rssb_parameter_t g_frag_run;
@@ -1258,19 +1262,23 @@ int main(int argc, char **argv)
else
{
g_frag_run.recv_bizman_lq = (MESA_lqueue_head*)calloc(1, g_frag_cfg.thread_num*sizeof(MESA_lqueue_head));
for(i=0;i<g_frag_cfg.thread_num;i++)
{
for(i=0;i<g_frag_cfg.thread_num;i++)
{
g_frag_run.recv_bizman_lq[i] = MESA_lqueue_create(1, 0);
}
for(i=0;i<g_frag_cfg.thread_num;i++)
{
if(-1 == create_pthread(bizman_recv_data_to_queue, (void*)i, g_frag_run.logger))
{
MESA_handle_runtime_log(g_frag_run.logger, RLOG_LV_FATAL, FRAG_REASSEMBLY_MODULE_NAME,
(char*)"[%s:%d] Thread bizman_recv_data_to_queue Create Failed." , __FILE__,__LINE__,FRAG_REASSEMBLY_MODULE_NAME);
return -1;
}
}
}
for(i=0;i<g_frag_cfg.thread_num;i++)
{
{
if(-1 == create_pthread(bizman_recv_data_from_queue,(void*)i,g_frag_run.logger))
{
MESA_handle_runtime_log(g_frag_run.logger, RLOG_LV_FATAL, FRAG_REASSEMBLY_MODULE_NAME,

4
src/main.h
View File

@@ -5,7 +5,7 @@
#include </usr/include/stdint.h>
#define APP_FUNC 0 //<2F>ֻ<EFBFBD>APP
#define VOIP_FUNC 0 //VOIP <20><>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ļ<EFBFBD><C4BC><EFBFBD>һ<EFBFBD><D2BB>
#define VOIP_FUNC 1 //VOIP <20><>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ļ<EFBFBD><C4BC><EFBFBD>һ<EFBFBD><D2BB>
#define PIC_FUNC 0 //PIC <20><>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ļ<EFBFBD><C4BC><EFBFBD>һ<EFBFBD><D2BB>
#include "MESA_list_queue.h"
@@ -62,11 +62,13 @@ typedef struct frag_rssb_parameter_s
MESA_htable_handle media_hash; //media_hash
MESA_timer_t* index_query_timer[MAX_THREAD_NUM];
char index_query_timer_on[MAX_THREAD_NUM];
/*<2A><>Դ<EFBFBD><D4B4><EFBFBD><EFBFBD>ҵ<EFBFBD><D2B5>*/
void* dedup_hd; //av_dedup_handle
void* dedup_logger; //av_dedup log
MESA_timer_t* multisrc_timer[MAX_THREAD_NUM];
char multisrc_timer_on[MAX_THREAD_NUM];
KafkaProducer* kafka_producer; //kafka

2
src/message.c
View File

@@ -409,7 +409,7 @@ void send_json_log(media_t* mdi, resp_checkresult_t* check_res)
memset(pbuf, 0, sizeof(pbuf));
snprintf(pbuf, sizeof(pbuf), "%" PRIu64 "", mdi->lastpkt_time - mdi->create_time);
cJSON_AddStringToObject(root, "duation", pbuf);
cJSON_AddStringToObject(root, "duration", pbuf);
if((NULL != mdi->sip_opt[SIP_FROM_OPT_INDEX])&&(mdi->sip_opt[SIP_FROM_OPT_INDEX]->opt_len > 0)&&(NULL != mdi->sip_opt[SIP_FROM_OPT_INDEX]->opt_value))
{