tango-tfe/plugin/business/decrypt-mirroring/src/mirror_stream.cpp

#include <stdio.h>   
#include <string.h>
#include <stdlib.h>
#include <unistd.h> 
#include <fcntl.h>
#include <errno.h>
#include <arpa/inet.h>
#include <net/if.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <netinet/tcp.h>
#include <netinet/ether.h>
#include <netpacket/packet.h>
#include <pthread.h>

#include <MESA/MESA_prof_load.h>
#include <MESA/MESA_handle_logger.h>
#include <MESA/field_stat2.h>

#include <mirror_stream.h>
#include <mirror_stream_inl.h>
#include <sendpkt-inl.h>

int g_deliver_version_VERSION_20180718;

struct deliver_globle_info g_deliver_globle_info;




int deliver_set_filestate2(int thread_seq,int colum_index,int value)
{
	if(thread_seq>=g_deliver_globle_info.comminfo.threadnum)
	{
		MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_FATAL,"deliver_set_filestate2","thread_seq:%d,thread_num:%d",thread_seq,g_deliver_globle_info.comminfo.threadnum);
	}

	g_deliver_globle_info.fs2info.column_value[thread_seq][colum_index]+=value;

	return 0;

}


int deliver_filestate2_init(char* filepath)
{	
	int i=0;
	int trans_switch=0;
	int value=1;
	unsigned int fs2_sport;
	char fs2_filename[DELIVER_CONF_MAXLEN]={0};
	char fs2_sip[DELIVER_CONF_MAXLEN]={0};

	MESA_load_profile_string_def((char*)filepath,(char*)DELIVER_CONF_MODE,(char*)"filestat2_filename",fs2_filename,DELIVER_CONF_MAXLEN,(char*)"./log/deliver_fs2.log");
	MESA_load_profile_string_def((char*)filepath,(char*)DELIVER_CONF_MODE,(char*)"filestat2_sip",fs2_sip,DELIVER_CONF_MAXLEN,(char*)"192.168.11.241");
	MESA_load_profile_uint_def((char*)filepath,(char*)DELIVER_CONF_MODE,(char*)"filestat2_sport",(unsigned int*)&fs2_sport,8125);
	MESA_load_profile_uint_def((char*)filepath,(char*)DELIVER_CONF_MODE,(char*)"filestat2_trans_switch",(unsigned int*)&trans_switch,0);

	g_deliver_globle_info.fs2info.handler=FS_create_handle();

	FS_set_para(g_deliver_globle_info.fs2info.handler, OUTPUT_DEVICE,fs2_filename, strlen(fs2_filename)+1);
	FS_set_para(g_deliver_globle_info.fs2info.handler, PRINT_MODE, &value, sizeof(value));
	FS_set_para(g_deliver_globle_info.fs2info.handler, STAT_CYCLE, &value, sizeof(value));
	FS_set_para(g_deliver_globle_info.fs2info.handler, CREATE_THREAD, &value, sizeof(value));
	FS_set_para(g_deliver_globle_info.fs2info.handler, APP_NAME, FS2_APPNAME, strlen(FS2_APPNAME)+1);

	if(trans_switch==1)
	{
		FS_set_para(g_deliver_globle_info.fs2info.handler, STATS_SERVER_IP, fs2_sip, strlen(fs2_sip)+1);
		FS_set_para(g_deliver_globle_info.fs2info.handler, STATS_SERVER_PORT,&fs2_sport,sizeof(int));
	}

	for(i=0;i<FS2_COLUMN_NUM;i++)
	{
		g_deliver_globle_info.fs2info.column_id[i]=FS_register(g_deliver_globle_info.fs2info.handler, FS_STYLE_FIELD, FS_CALC_CURRENT,g_deliver_globle_info.fs2_name[i]);
	}


	FS_start(g_deliver_globle_info.fs2info.handler);

	return 0;
}


void* deliver_filestat2(void* arg)
{
	int i=0;
	int j=0;

	char* filepath=(char*)arg;

	unsigned long long column_value[FS2_COLUMN_NUM];

	deliver_filestate2_init(filepath);

	while(1)
	{
		for(i=0;i<FS2_COLUMN_NUM;i++)
		{
			column_value[i]=0;
		
			for(j=0;j<g_deliver_globle_info.comminfo.threadnum;j++)
			{
				column_value[i]+=g_deliver_globle_info.fs2info.column_value[j][i];
			}
		
			FS_operate(g_deliver_globle_info.fs2info.handler,g_deliver_globle_info.fs2info.column_id[i], 0,FS_OP_SET,column_value[i]);
		}

		sleep(1);
	}

	return NULL;
}



int deliver_debug_log_v6(int level,char* module,struct deliver_pme_info* pmeinfo,int flag,int recv_len)
{	
	struct deliver_session_info* session_info=&(pmeinfo->session_info);
	struct deliver_pkt_info* pkt_info=&(pmeinfo->pkt_info);

	const struct tfe_stream_addr_tuple4_v6* tuple4_v6 = pmeinfo->addr_info->tuple4_v6;
	
	char saddr_v6[INET6_ADDRSTRLEN ]={0};
	char daddr_v6[INET6_ADDRSTRLEN ]={0};
	unsigned short sport=ntohs(tuple4_v6->source);
	unsigned short dport=ntohs(tuple4_v6->dest);

	inet_ntop(AF_INET6, (void *)&(tuple4_v6->saddr), saddr_v6, INET6_ADDRSTRLEN);
	inet_ntop(AF_INET6, (void *)&(tuple4_v6->daddr), daddr_v6, INET6_ADDRSTRLEN);

	switch(flag)
	{
		case DELIVER_FLAG_SENDPKT:			
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,level,module,"addr:%s,%d,%s,%d,len:%d,dir:%d,seq:%u,ack:%u,ipid:%d,win:%d,flag:%d",
						saddr_v6,sport,daddr_v6,dport,pkt_info->len,pkt_info->dir,
						pkt_info->seq,pkt_info->ack,pkt_info->ipid,pkt_info->win,pkt_info->flag);
			break;
		case DELIVER_FLAG_RECVPKT:
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,level,module,"addr:%s,%d,%s,%d,dir:%d,recv_len:%d",
									saddr_v6,sport,daddr_v6,dport,pkt_info->dir,recv_len);
			break;
		case DELIVER_FLAG_ENT:
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,level,module,"addr:%s,%d,%s,%d,recvpkt:%d,recvbyte:%lld,sendpkt:%d,sendbytes:%lld",
									saddr_v6,sport,daddr_v6,dport,session_info->recv_pkt,session_info->recv_byte,
									session_info->send_pkt,session_info->recv_byte);
			break;
	}


	return 0;
	
}



int deliver_debug_log_v4(int level,char* module,struct deliver_pme_info* pmeinfo,int flag,int recv_len)
{	
	struct deliver_session_info* session_info=&(pmeinfo->session_info);
	struct deliver_pkt_info* pkt_info=&(pmeinfo->pkt_info);


	const struct tfe_stream_addr_tuple4_v4* tuple4_v4 = pmeinfo->addr_info->tuple4_v4;
	
	char saddr_v4[INET_ADDRSTRLEN]={0};
	char daddr_v4[INET_ADDRSTRLEN]={0};
	unsigned short sport=ntohs(tuple4_v4->source);
	unsigned short dport=ntohs(tuple4_v4->dest);

	inet_ntop(AF_INET, (void *)&(tuple4_v4->saddr), saddr_v4, INET_ADDRSTRLEN);
	inet_ntop(AF_INET, (void *)&(tuple4_v4->daddr), daddr_v4, INET_ADDRSTRLEN);

	switch(flag)
	{
		case DELIVER_FLAG_SENDPKT:			
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,level,module,"addr:%s,%d,%s,%d,len:%d,dir:%d,seq:%u,ack:%u,ipid:%d,win:%d,flag:%d",
						saddr_v4,sport,daddr_v4,dport,pkt_info->len,pkt_info->dir,
						pkt_info->seq,pkt_info->ack,pkt_info->ipid,pkt_info->win,pkt_info->flag);
			break;
		case DELIVER_FLAG_RECVPKT:
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,level,module,"addr:%s,%d,%s,%d,dir:%d,recv_len:%d",
									saddr_v4,sport,daddr_v4,dport,pkt_info->dir,recv_len);
			break;
		case DELIVER_FLAG_ENT:
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,level,module,"addr:%s,%d,%s,%d,recvpkt:%d,recvbyte:%lld,sendpkt:%d,sendbytes:%lld",
									saddr_v4,sport,daddr_v4,dport,session_info->recv_pkt,session_info->recv_byte,
									session_info->send_pkt,session_info->recv_byte);
			break;
	}

	return 0;
	
}


int deliver_sendpkt_ether(int thread_seq,int buflen,unsigned char* buf,unsigned char* dmac)
{
	int ret=0;

	if(-1==ioctl(g_deliver_globle_info.sendinfo.send_socket[thread_seq],SIOCGIFINDEX,&(g_deliver_globle_info.sendinfo.ifr)))
	{
		MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_FATAL,DELIVER_MODULE_INIT,"get if index error:%d,%s,name:%d",errno,strerror(errno),g_deliver_globle_info.sendinfo.senddevice);
		return -1;
	}

//	struct sockaddr_ll addr={0};
	struct sockaddr_ll addr;
	addr.sll_family=AF_PACKET;
	addr.sll_halen=ETHER_ADDR_LEN;
	addr.sll_ifindex=g_deliver_globle_info.sendinfo.ifr.ifr_ifindex;
	addr.sll_protocol=htons(ETH_P_IP);
	memcpy(addr.sll_addr,dmac,ETHER_ADDR_LEN);


	if(ioctl(g_deliver_globle_info.sendinfo.send_socket[thread_seq],SIOCGIFHWADDR,&(g_deliver_globle_info.sendinfo.ifr))==-1)
	{
		return -1;
	}

	ret=sendto(g_deliver_globle_info.sendinfo.send_socket[thread_seq],buf,buflen,0,(struct sockaddr*)&addr,sizeof(addr));
	if(ret<0)
	{
		deliver_set_filestate2(thread_seq,FS2_COLUME_ERROR,1);
		MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_FATAL,DELIVER_MODULE_SENDPKT,"sendto() error,errno:%d,msg:%s!",errno,strerror(errno));
		return ret;
	}

	
	deliver_set_filestate2(thread_seq,FS2_COLUME_SENDPKT,1);
	deliver_set_filestate2(thread_seq,FS2_COLUME_SENDBYTE,buflen-14-20-20);

	return ret;

}


int deliver_init_pmeinfo(const struct tfe_stream_addr* addr,void** pme)
{
//TODO:choose dst mac
	int i=deliver_rand()%(g_deliver_globle_info.sendinfo.receiver_num);

	struct deliver_pme_info* pmeinfo=(struct deliver_pme_info*)malloc(sizeof(struct deliver_pme_info));
	memset(pmeinfo,0,sizeof(struct deliver_pme_info));
	pmeinfo->addr_info = addr;
	memcpy(pmeinfo->dst_macaddr,g_deliver_globle_info.sendinfo.receiver_info[i].dst_macaddr,DELIVER_MACADDR_LEN);

	*pme=pmeinfo;

	return 0;
}


int deliver_send_v6(int thread_seq,struct deliver_pme_info* pmeinfo,int payload_len,const unsigned char* payload)
{
	int offset = 0;
	unsigned short eth_type=0x0800;

	
	int cur_dir=pmeinfo->pkt_info.dir;
	const struct tfe_stream_addr* addr_info = pmeinfo->addr_info;
	struct deliver_pkt_info* curpkt_info=&(pmeinfo->pkt_info);

	offset = sizeof(struct mesa_ethernet_hdr);

	if(cur_dir==CONN_DIR_DOWNSTREAM)
	{
		deliver_build_tcp(ntohs(addr_info->tuple4_v6->source), ntohs(addr_info->tuple4_v6->dest), 
							curpkt_info->seq, curpkt_info->ack,curpkt_info->flag, curpkt_info->win, 0,
							payload,payload_len, 
							g_deliver_globle_info.sendbuf[thread_seq]+offset+sizeof(struct mesa_ip6_hdr));
		
		deliver_build_ipv6(0, 0, payload_len + sizeof(struct mesa_tcp_hdr), IPPROTO_TCP, curpkt_info->ttl, 
							&(addr_info->tuple4_v6->saddr),&(addr_info->tuple4_v6->daddr),NULL, 0, 
							g_deliver_globle_info.sendbuf[thread_seq]+offset);
		
	}
	else
	{
		deliver_build_tcp(ntohs(addr_info->tuple4_v6->dest), ntohs(addr_info->tuple4_v6->source), 
							curpkt_info->seq, curpkt_info->ack,curpkt_info->flag, curpkt_info->win, 0,
							payload,payload_len, 
							g_deliver_globle_info.sendbuf[thread_seq]+offset+sizeof(struct mesa_ip6_hdr));
		
		deliver_build_ipv6(0, 0, payload_len + sizeof(struct mesa_tcp_hdr), IPPROTO_TCP, curpkt_info->ttl, 
							&(addr_info->tuple4_v6->daddr),&(addr_info->tuple4_v6->saddr),NULL, 0, 
							g_deliver_globle_info.sendbuf[thread_seq]+offset);
	}

	
	


	deliver_do_checksum(g_deliver_globle_info.sendbuf[thread_seq]+offset, IPPROTO_TCP, SENDPACKET_TCP_H+payload_len);
	
//	deliver_do_checksum(g_deliver_globle_info.sendbuf[thread_seq]+offset, IPPROTO_IP, SENDPACKET_IP_H);


	deliver_build_ethernet((unsigned char*)(pmeinfo->dst_macaddr),(unsigned char*)(g_deliver_globle_info.sendinfo.src_macaddr),
							eth_type,NULL,0,(unsigned char*)g_deliver_globle_info.sendbuf[thread_seq]);
	
	deliver_sendpkt_ether(thread_seq,SENDPACKET_TCP_H+SENDPACKET_IP_H+SENDPACKET_ETH_H+payload_len,
						g_deliver_globle_info.sendbuf[thread_seq],pmeinfo->dst_macaddr);


	pmeinfo->session_info.send_pkt++;
	pmeinfo->session_info.send_byte+=payload_len;
	
	deliver_debug_log_v6(RLOG_LV_DEBUG,(char*)DELIVER_SENDPKT_DEBUG,pmeinfo,DELIVER_FLAG_SENDPKT,0);

	return 0;

	
}


int deliver_send_v4(int thread_seq,struct deliver_pme_info* pmeinfo,int payload_len,const unsigned char* payload)
{
	int offset = 0;
	unsigned short eth_type=0x0800;

	
	int cur_dir=pmeinfo->pkt_info.dir;
	const struct tfe_stream_addr* addr_info = pmeinfo->addr_info;
	struct deliver_pkt_info* curpkt_info=&(pmeinfo->pkt_info);

	offset = sizeof(struct mesa_ethernet_hdr);

	if(cur_dir==CONN_DIR_DOWNSTREAM)
	{
		deliver_build_tcp(ntohs(addr_info->tuple4_v4->source), ntohs(addr_info->tuple4_v4->dest), 
						curpkt_info->seq, curpkt_info->ack,curpkt_info->flag, curpkt_info->win, 0,
						payload,payload_len, 
						g_deliver_globle_info.sendbuf[thread_seq]+offset+sizeof(struct mesa_ip4_hdr));
						
		deliver_build_ipv4(SENDPACKET_TCP_H+payload_len, 0, curpkt_info->ipid, 0, 64, IPPROTO_TCP, 
						addr_info->tuple4_v4->saddr.s_addr,addr_info->tuple4_v4->daddr.s_addr, NULL, 0, 
						g_deliver_globle_info.sendbuf[thread_seq]+offset);
	}
	else
	{
		deliver_build_tcp(ntohs(addr_info->tuple4_v4->dest), ntohs(addr_info->tuple4_v4->source), 
						curpkt_info->seq, curpkt_info->ack,curpkt_info->flag, curpkt_info->win, 0,
						payload,payload_len, 
						g_deliver_globle_info.sendbuf[thread_seq]+offset+sizeof(struct mesa_ip4_hdr));
						
		deliver_build_ipv4(SENDPACKET_TCP_H+payload_len, 0, curpkt_info->ipid, 0, 64, IPPROTO_TCP, 
						addr_info->tuple4_v4->daddr.s_addr,addr_info->tuple4_v4->saddr.s_addr, NULL, 0, 
						g_deliver_globle_info.sendbuf[thread_seq]+offset);
	}
	

	deliver_do_checksum(g_deliver_globle_info.sendbuf[thread_seq]+offset, IPPROTO_TCP, SENDPACKET_TCP_H+payload_len);
	
	deliver_do_checksum(g_deliver_globle_info.sendbuf[thread_seq]+offset, IPPROTO_IP, SENDPACKET_IP_H);


	deliver_build_ethernet((unsigned char*)(pmeinfo->dst_macaddr),(unsigned char*)(g_deliver_globle_info.sendinfo.src_macaddr),
							eth_type,NULL,0,(unsigned char*)g_deliver_globle_info.sendbuf[thread_seq]);
	
	deliver_sendpkt_ether(thread_seq,SENDPACKET_TCP_H+SENDPACKET_IP_H+SENDPACKET_ETH_H+payload_len,
						g_deliver_globle_info.sendbuf[thread_seq],pmeinfo->dst_macaddr);


	pmeinfo->session_info.send_pkt++;
	pmeinfo->session_info.send_byte+=payload_len;
	
	deliver_debug_log_v4(RLOG_LV_DEBUG,(char*)DELIVER_SENDPKT_DEBUG,pmeinfo,DELIVER_FLAG_SENDPKT,0);


	return 0;

	
}

int deliver_send_syn(int thread_seq,struct deliver_pme_info* pmeinfo)
{
	pmeinfo->pkt_info.dir=CONN_DIR_DOWNSTREAM;
	pmeinfo->pkt_info.len=0;
	pmeinfo->pkt_info.seq= deliver_rand();
	pmeinfo->pkt_info.ack=0;
	pmeinfo->pkt_info.flag=TH_SYN;
	pmeinfo->pkt_info.win = deliver_rand_range(1460, 65500);
	pmeinfo->pkt_info.ipid = deliver_rand() % 65535;
	pmeinfo->pkt_info.ttl=deliver_rand_range(32,65);
	if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V4)
	{
		deliver_send_v4(thread_seq,pmeinfo,0,NULL);		
		deliver_debug_log_v4(RLOG_LV_INFO,(char*)DELIVER_SENDPKT_START,pmeinfo,DELIVER_FLAG_SENDPKT,0);
	}
	else if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V6)
	{
		deliver_send_v6(thread_seq,pmeinfo,0,NULL);		
		deliver_debug_log_v6(RLOG_LV_INFO,(char*)DELIVER_SENDPKT_START,pmeinfo,DELIVER_FLAG_SENDPKT,0);
	}
	else
	{
		//TODO
	}

	return 0;
}

int deliver_send_syn_ack(int thread_seq,struct deliver_pme_info* pmeinfo)
{
	pmeinfo->pkt_info.dir=CONN_DIR_UPSTREAM;
	pmeinfo->pkt_info.len=0;
	pmeinfo->pkt_info.ack=pmeinfo->pkt_info.seq+1;
	pmeinfo->pkt_info.seq= deliver_rand();
	pmeinfo->pkt_info.flag=TH_SYN|TH_ACK;

	if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V4)
	{
		deliver_send_v4(thread_seq,pmeinfo,0,NULL);
	}
	else if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V6)
	{
		deliver_send_v6(thread_seq,pmeinfo,0,NULL);
	}
	else
	{
//TODO
	}

	return 0;
}



int deliver_send_ack(int thread_seq,struct deliver_pme_info* pmeinfo)
{
	unsigned int ack_tmp=pmeinfo->pkt_info.ack;
	
	pmeinfo->pkt_info.dir=CONN_DIR_DOWNSTREAM;
	pmeinfo->pkt_info.len=0;
	pmeinfo->pkt_info.ack=pmeinfo->pkt_info.seq+1;
	pmeinfo->pkt_info.seq= ack_tmp;
	pmeinfo->pkt_info.flag=TH_ACK;

	
	if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V4)
	{
		deliver_send_v4(thread_seq,pmeinfo,0,NULL);
	}
	else if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V6)
	{
		deliver_send_v6(thread_seq,pmeinfo,0,NULL);
	}
	else
	{
//todo
	}

	return 0;
}


int deliver_set_pktinfo(struct deliver_pme_info* pmeinfo,int flag, int cur_dir,int payload_len)
{
	struct deliver_pkt_info last_pkt_info;
	memcpy((void*)&last_pkt_info,(void*)&(pmeinfo->pkt_info),sizeof(struct deliver_pkt_info));

	pmeinfo->pkt_info.dir=cur_dir;
	pmeinfo->pkt_info.len=payload_len;
	pmeinfo->pkt_info.flag=flag;
	

	if(cur_dir==last_pkt_info.dir)
	{
		pmeinfo->pkt_info.seq=last_pkt_info.seq+last_pkt_info.len;
		pmeinfo->pkt_info.ack=last_pkt_info.ack;
	}
	else
	{
		pmeinfo->pkt_info.seq=last_pkt_info.ack;
		pmeinfo->pkt_info.ack=last_pkt_info.seq+last_pkt_info.len;
	}

	return 0;

}

int deliver_send_rst(int thread_seq,struct deliver_pme_info* pmeinfo)
{
	deliver_set_pktinfo(pmeinfo,TH_RST,CONN_DIR_DOWNSTREAM,0);
	if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V4)
	{
		deliver_send_v4(thread_seq,pmeinfo,0,NULL);
	}
	else if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V6)
	{
		deliver_send_v6(thread_seq,pmeinfo,0,NULL);
	}
	else
	{
//TODO
	}


	deliver_set_pktinfo(pmeinfo,TH_RST,CONN_DIR_UPSTREAM,0);
	if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V4)
	{
		deliver_send_v4(thread_seq,pmeinfo,0,NULL);
		deliver_debug_log_v4(RLOG_LV_INFO,(char*)DELIVER_SENDPKT_END,pmeinfo,DELIVER_FLAG_ENT,0);
	}
	else if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V6)
	{
		deliver_send_v6(thread_seq,pmeinfo,0,NULL);
		deliver_debug_log_v6(RLOG_LV_INFO,(char*)DELIVER_SENDPKT_END,pmeinfo,DELIVER_FLAG_ENT,0);
	}
	else
	{
//TODO
	}

	return 0;
}



int mirror_stream_open(int thread_id, const struct tfe_stream_addr* addr,void** pme)
{
	if(g_deliver_globle_info.deliver_switch == 0)
	{
		MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_INFO,DELIVER_MODULE_INIT,"mirror_stream switch is 0,not run....");
		return -1;
	}

	struct deliver_pme_info* pmeinfo=NULL;

	deliver_init_pmeinfo(addr,pme);
	pmeinfo=(struct deliver_pme_info*)*pme;
	
	deliver_send_syn(thread_id,pmeinfo);	
	deliver_send_syn_ack(thread_id,pmeinfo);	
	deliver_send_ack(thread_id,pmeinfo);	

	return 0;

}

int mirror_stream_write(int cur_dir,const unsigned char * data, size_t len, void** pme,int thread_id)
{
	if(g_deliver_globle_info.deliver_switch == 0)
	{
		return -1;
	}

	int i=0;
	const unsigned char* payload=data;
	int payload_len=0;
	int remain_len=len;
	int max_payloadlen_per=0;
	int pkt_num=0;
//	int pkt_num=(len/(g_deliver_globle_info.sendinfo.mtu))+1;
	struct deliver_pme_info* pmeinfo=(struct deliver_pme_info*)*pme;

	
	deliver_set_filestate2(thread_id,FS2_COLUME_RECVPKT,1);
	deliver_set_filestate2(thread_id,FS2_COLUME_RECVBYTE,len);

	pmeinfo->session_info.recv_pkt++;
	pmeinfo->session_info.recv_byte+=len;
	if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V4)
	{
		deliver_debug_log_v4(RLOG_LV_DEBUG,(char*)DELIVER_RECVPKT_DEBUG,pmeinfo,DELIVER_FLAG_RECVPKT,len);
		max_payloadlen_per=g_deliver_globle_info.sendinfo.mtu-sizeof(struct mesa_tcp_hdr)-sizeof(struct mesa_ip4_hdr);
	}
	else if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V6)
	{
		deliver_debug_log_v6(RLOG_LV_DEBUG,(char*)DELIVER_RECVPKT_DEBUG,pmeinfo,DELIVER_FLAG_RECVPKT,len);
		max_payloadlen_per=g_deliver_globle_info.sendinfo.mtu-sizeof(struct mesa_tcp_hdr)-sizeof(struct mesa_ip6_hdr);
	}
	else
	{
		//TODO
	}

	pkt_num=(len/(max_payloadlen_per))+1;
	
	for(i=0;i<pkt_num;i++)
	{
/*
		payload_len=remain_len<(g_deliver_globle_info.sendinfo.mtu)?remain_len:(g_deliver_globle_info.sendinfo.mtu);
		payload=data+(len-remain_len);
		remain_len-=g_deliver_globle_info.sendinfo.mtu;
*/	
		payload_len=remain_len<max_payloadlen_per?remain_len:max_payloadlen_per;
		payload=data+(len-remain_len);
		remain_len-=max_payloadlen_per;

		deliver_set_pktinfo(pmeinfo,TH_ACK,cur_dir,payload_len);
		if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V4)
		{
			deliver_send_v4(thread_id,pmeinfo,payload_len,payload);
		}
		else if(pmeinfo->addr_info->addrtype==TFE_ADDR_STREAM_TUPLE4_V6)
		{
			deliver_send_v6(thread_id,pmeinfo,payload_len,payload);
		}
		else
		{
//TODO
		}
	}

	return 0;
}


void mirror_stream_close(void** pme, int thread_id)
{
	if(g_deliver_globle_info.deliver_switch == 0)
	{
		return;
	}


	struct deliver_pme_info* pmeinfo=(struct deliver_pme_info*)*pme;
	
	deliver_send_rst(thread_id,pmeinfo);


	if(*pme!=NULL)
	{
		free(*pme);
		*pme=NULL;
	}
	return;
}


int deliver_device_init()
{
	char* if_name=g_deliver_globle_info.sendinfo.senddevice;

//init socket
	size_t ifname_len=strlen(if_name);
	if(ifname_len<sizeof(g_deliver_globle_info.sendinfo.ifr.ifr_name))
	{
		memset(g_deliver_globle_info.sendinfo.ifr.ifr_name,0,IFNAMSIZ);
		memcpy(g_deliver_globle_info.sendinfo.ifr.ifr_name,if_name,ifname_len);
	}
	else
	{
		MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_FATAL,DELIVER_MODULE_INIT,"interface name :%s is too long\n",if_name);
		return -1;
	}

	deliver_get_dev_mac(g_deliver_globle_info.sendinfo.senddevice,g_deliver_globle_info.sendinfo.src_macaddr);
	
	return 0;

}

int deliver_profile_init(const char* filepath,int* logger_level,char* logger_filepath)
{
	int i=1;
	char mac_addr_str[DELIVER_MACADDR_STR_LEN];
	char receiver_mac_name[DELIVER_CARDNAME_LEN]={0};

//switch
	MESA_load_profile_int_def(filepath,(char*)DELIVER_CONF_MODE,(char*)"mirror_switch",&(g_deliver_globle_info.deliver_switch),0);


//runtime log	
	MESA_load_profile_int_def(filepath,(char*)DELIVER_CONF_MODE,(char*)"logger_level",logger_level,RLOG_LV_INFO);
	MESA_load_profile_string_def(filepath,(char*)DELIVER_CONF_MODE,(char*)"logger_filepath",logger_filepath,DELIVER_CONF_MAXLEN,"./log/deliver.log");

//sendpkt info
	MESA_load_profile_int_def(filepath,(char*)DELIVER_CONF_MODE,(char*)"mtu",&(g_deliver_globle_info.sendinfo.mtu),DELIVER_DEFAULT_MTU);
	MESA_load_profile_int_def(filepath,(char*)DELIVER_CONF_MODE,(char*)"receiver_num",&(g_deliver_globle_info.sendinfo.receiver_num),1);
	MESA_load_profile_string_nodef(filepath,(char*)DELIVER_CONF_MODE,(char*)"senddevice",g_deliver_globle_info.sendinfo.senddevice,DELIVER_CARDNAME_LEN);

	g_deliver_globle_info.sendinfo.receiver_info=(struct deliver_recver_info*)ALLOC(struct deliver_recver_info,g_deliver_globle_info.sendinfo.receiver_num);
	for(i=1;i<=g_deliver_globle_info.sendinfo.receiver_num;i++)
	{
		memset(mac_addr_str,0,DELIVER_MACADDR_STR_LEN);
	
		memset(receiver_mac_name,0,DELIVER_CARDNAME_LEN);
		sprintf(receiver_mac_name,"%s%d",DELIVER_RCV_MAC,i);

		MESA_load_profile_string_nodef(filepath,(char*)DELIVER_CONF_MODE,(char*)receiver_mac_name,(char*)mac_addr_str,DELIVER_MACADDR_STR_LEN);
		
		if(deliver_mac_pton(mac_addr_str, ':', (char *)(g_deliver_globle_info.sendinfo.receiver_info[i-1].dst_macaddr)) < 0)
		{
			printf("error, receiver %d mac:%s is not correct, for example:00:11:22:33:44:55\n", i,mac_addr_str);
			return -1;
		}
	}


	return 0;
}

int deliver_socket_error(int n)
{
	for(int i=0;i<n;i++)
	{
		close(g_deliver_globle_info.sendinfo.send_socket[i]);
		g_deliver_globle_info.sendinfo.send_socket[i]=0;
	}

	return 0;
}

int deliver_init_log()
{
	int i=0;
	int j=0;
	
	char mac_str[DELIVER_MACADDR_STR_LEN]={0};
	
	for(j = 0; j < 6; j++)
	{
		sprintf(mac_str+3*j, "%02x:", (unsigned char)g_deliver_globle_info.sendinfo.src_macaddr[j]);
	}
	mac_str[DELIVER_MACADDR_STR_LEN]=0;
	MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_FATAL,DELIVER_MODULE_INIT,"mtu:%d,receiver_num:%d,senddevice:%s,mac:%s",
					g_deliver_globle_info.sendinfo.mtu,g_deliver_globle_info.sendinfo.receiver_num,g_deliver_globle_info.sendinfo.senddevice,mac_str);

	for(i=0;i<g_deliver_globle_info.sendinfo.receiver_num;i++)
	{
		
		for(j = 0; j < 6; j++)
		{
			sprintf(mac_str+3*j, "%02x:", (unsigned char)g_deliver_globle_info.sendinfo.receiver_info[i].dst_macaddr[j]);
		}

		mac_str[DELIVER_MACADDR_STR_LEN]=0;
		MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_FATAL,DELIVER_MODULE_INIT,"reveiver_mac:%s",mac_str);
	}

	return 0;

}

int mirror_stream_init(int thread_num, const char* filepath)
{
	int i=0;
	int flag=0;
	int logger_level;
	char logger_filepath[DELIVER_CONF_MAXLEN]={0};

	pthread_t pid_deliver_filestat2;

	g_deliver_globle_info.comminfo.threadnum=thread_num;
	g_deliver_globle_info.fs2_name[FS2_COLUME_RECVPKT] = (char*)"RECV_PKTS";
	g_deliver_globle_info.fs2_name[FS2_COLUME_RECVBYTE] = (char*)"RECV_BYTES";
	g_deliver_globle_info.fs2_name[FS2_COLUME_SENDPKT] = (char*)"SEND_PKTS";
	g_deliver_globle_info.fs2_name[FS2_COLUME_SENDBYTE] = (char*)"SEND_BYTES";
	g_deliver_globle_info.fs2_name[FS2_COLUME_ERROR] = (char*)"ERROR_PKTS";

//profile
	deliver_profile_init(filepath,&logger_level,logger_filepath);


//init runtime log
	g_deliver_globle_info.comminfo.logger=MESA_create_runtime_log_handle(logger_filepath,logger_level);
	if(g_deliver_globle_info.comminfo.logger==NULL)
	{
		printf("MESA_create_runtime_log_handle() error!exit...\n");
		return -1;
	}
	
//socket init
	g_deliver_globle_info.sendinfo.thread_num=thread_num;
	g_deliver_globle_info.sendinfo.send_socket=(int*)malloc(g_deliver_globle_info.sendinfo.thread_num*sizeof(int));
	for(i=0;i<g_deliver_globle_info.sendinfo.thread_num;i++)
	{
		g_deliver_globle_info.sendbuf[i] = (unsigned char*)malloc(g_deliver_globle_info.sendinfo.mtu + sizeof(struct mesa_ethernet_hdr));
	
		g_deliver_globle_info.sendinfo.send_socket[i]=socket(AF_PACKET,SOCK_RAW,htons(ETH_P_IP));
		if((g_deliver_globle_info.sendinfo.send_socket[i]<0) ||(g_deliver_globle_info.sendbuf[i]==NULL))
		{
			deliver_socket_error(i);
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger, RLOG_LV_FATAL,DELIVER_MODULE_INIT,"ipv4_raw_socket error,i:%d",i);
			return -1;
		}
		flag= fcntl(g_deliver_globle_info.sendinfo.send_socket[i], F_GETFL, 0);
		if(flag<0)
		{
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger, RLOG_LV_FATAL,DELIVER_MODULE_INIT,"fcntl():getfl error,errno is:%d,%s",errno,strerror(errno));
		}
		if( fcntl( g_deliver_globle_info.sendinfo.send_socket[i], F_SETFL, flag|O_NONBLOCK ) < 0 )
		{
			MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger, RLOG_LV_FATAL,DELIVER_MODULE_INIT,"fcntl():setfl error,errno is:%d,%s",errno,strerror(errno));
		}
	}
	
	pthread_create(&pid_deliver_filestat2,NULL,deliver_filestat2,(void*)filepath);

	deliver_device_init();
	
	deliver_init_log();

	MESA_handle_runtime_log(g_deliver_globle_info.comminfo.logger,RLOG_LV_FATAL,DELIVER_MODULE_INIT,"thread_num:%d,filepath:%s,deliver.so init sucess!",thread_num,filepath);

	return 0;

}