/*
 *  http_url_discovery
 *  author:yangwei
 *  create time:2014-8-18
 */



#include "ntc_http_collect.h"


#include <stdio.h>
#include <time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/ioctl.h>
#include <net/if.h>
#include <unistd.h>
#include <sys/prctl.h>
#include <sys/ioctl.h>
#include <dlfcn.h>

#include "MESA_prof_load.h"
#include "MESA_handle_logger.h"
#include "http.h"
#include "cJSON.h"
#include "MESA_htable.h"
#include "field_stat2.h"
#include "rdkafka.h"




#define LOCAL_IP "127.0.0.1"

#define PLUGIN_NAME "NTC_HTTP_COLLECT"
#define PROFILE_PATH "./t1conf/main.conf"
#define FILTER_PATH "./t1conf/http_url_filter.conf"

#ifndef MAX_BUF_LEN
#define MAX_BUF_LEN 4096
#endif

#ifndef MAX_THREAD_NUM
#define MAX_THREAD_NUM 128
#endif


#define MIN(a, b)  (((a) < (b)) ? (a) : (b))

#ifdef __cplusplus
extern "C"
{
#endif

#define GIT_VERSION_CATTER(v) __attribute__((__used__)) const char * GIT_VERSION_##v = NULL
#define GIT_VERSION_EXPEND(v) GIT_VERSION_CATTER(v)

/* VERSION TAG */
#ifdef GIT_VERSION
GIT_VERSION_EXPEND(GIT_VERSION);
#else
static __attribute__((__used__)) const char * GIT_VERSION_UNKNOWN = NULL;
#endif
#undef GIT_VERSION_CATTER
#undef GIT_VERSION_EXPEND

#ifdef __cplusplus
}
#endif

int NTC_HTTP_COLLECT_VERSION_20180910 = 1;

g_http_url_discovery_item g_item;

extern int g_iThreadNum;
extern long long  g_CurrentTime;

MESA_htable_handle g_url_filter_hash_handle;

rd_kafka_t *g_multi_kafka_producer;

rd_kafka_t *g_ntc_http_collect_kafka_handle = NULL;
rd_kafka_topic_t *g_ntc_http_collect_kafka_topic = NULL;

static unsigned int get_ip_by_eth_name(const char *ifname)
{
	int sockfd;
	struct ifreq ifr;
	unsigned int ip;

	sockfd = socket(AF_INET, SOCK_DGRAM, 0);
	if (-1 == sockfd) {
		goto error;
	}

	strcpy(ifr.ifr_name,ifname);
	if (ioctl(sockfd, SIOCGIFADDR, &ifr) < 0) {
		goto error;
	}

	ip = ((struct sockaddr_in*)&(ifr.ifr_addr))->sin_addr.s_addr;
	close(sockfd);
	return ip;

error:
	close(sockfd);
	return INADDR_NONE;
}

typedef struct store_hash_data_t
{
	char *url_buf;
	char *refer_buf;
	unsigned int url_len;
	unsigned int refer_len;
	long long store_time;
}store_hash_data_t;

static void store_data_free(void *data)
{
	if(data == NULL)
		return;
	store_hash_data_t *p_hash_data = (store_hash_data_t *)data;
	if(p_hash_data->url_buf != NULL)
	{
		free(p_hash_data->url_buf);
		p_hash_data->url_buf = NULL;
	}
	if(p_hash_data->refer_buf != NULL)
	{
		free(p_hash_data->refer_buf);
		p_hash_data->refer_buf = NULL;
	}
	free(data);
	data = NULL;
}


static int append_http_query_key(struct streaminfo *a_tcp,int http_seq, char *out_buff, int buff_len)
{
	if(a_tcp == NULL || http_seq <= 0 || out_buff == NULL || buff_len <= 0)
		return -1;
    unsigned char proxy_type = 0;
    char tuple4_addr[1024];
    memset(tuple4_addr, 0, sizeof(tuple4_addr));
    unsigned short tuple4_addr_len= 0;
    int tcp_seq = 0, ret = 0;
    int tcp_seq_len = sizeof(tcp_seq);
	unsigned short tunnel_type;
    int tunnel_type_len = sizeof(short);
	
    if(a_tcp->pfather != NULL)
    {
        if(a_tcp->pfather->type == STREAM_TYPE_HTTP_PROXY)
            proxy_type = 1;
    }

    ret = MESA_get_stream_opt(a_tcp, MSO_TCP_ISN_C2S, &tcp_seq, &tcp_seq_len);
    if(ret != 0)return -1;


    ret = MESA_get_stream_opt(a_tcp, MSO_STREAM_TUNNEL_TYPE, &tunnel_type, &tunnel_type_len);
    if(ret != 0)return -1;	

    if(STREAM_TUNNLE_NON ==  tunnel_type)
    {
    	const char * addr = printaddr(&a_tcp->addr, a_tcp->threadnum);
		if(addr == NULL)
			return -1;
		memcpy(tuple4_addr, addr, strlen(addr));
		tuple4_addr_len = strlen(addr);
    }
    else
    {
            tuple4_addr_len = stream_addr_list_ntop(a_tcp, tuple4_addr, sizeof(tuple4_addr));
    }
    if(tuple4_addr_len <= 0)
      return -1;
	snprintf(out_buff, buff_len, "%s#%d#%u#%u", tuple4_addr, http_seq, proxy_type, (unsigned int)tcp_seq);
	return 0;
}

static long hash_search_filter_cb_func(void *data, const uchar *key, uint size, void *user_arg)
{
	if(data == NULL)
		return 0;
	else
		return 1;
}


static int check_url_filter(char *url, int url_len)
{
	if(url == NULL || url_len <= 0)
		return 1;
	char *p_pos = url+url_len-1;
	int i = 0;
	long ret = 0;
	for(i = 0; i < url_len-1; i++)
	{
		if(*p_pos == '.')
			break;
		p_pos-=1;
	}
	if((i == (url_len-1)) && (*p_pos!='.'))
		return 0;

	MESA_htable_search_cb(g_url_filter_hash_handle, (uchar *)p_pos, i+1,
			(hash_cb_fun_t *)hash_search_filter_cb_func, NULL, &ret);
	return (int)ret;
}

static int  form_url_to_domain(const char  *url, char  *domain)
{
    int  i = 0;
    const char  *ptr = NULL; //define the  first  begin
    ptr = strstr(url, "://");

    if(ptr != NULL)
        ptr = ptr + strlen("://");
    else
        ptr = url;

    for(; *ptr != '\0' && *ptr != '/' && *ptr != ':' ; ptr++)
    {
        domain[i++] = *ptr;
    }

    domain[i] = '\0';
    return i;
}

static int ntc_addStreamInfo_to_jsonObj(cJSON *json_obj, const struct streaminfo *a_stream)
{
    const char *addr_proto = NULL;
    const char *null_addr = "0";
    short null_port = 0;
    unsigned short tunnel_type = 0;
    char nest_addr_buf[1024];
    int tunnel_type_size = sizeof(tunnel_type);
    const struct layer_addr *addr = NULL;
    char src_ip_str[128] = {0}, dst_ip_str[128] = {0};

    cJSON_AddNumberToObject(json_obj, "stream_dir", a_stream->dir);

    addr = &(a_stream->addr);
    switch (addr->addrtype)
    {
        case ADDR_TYPE_IPV4:
        case __ADDR_TYPE_IP_PAIR_V4:
            inet_ntop(AF_INET, &addr->ipv4->saddr, src_ip_str, sizeof(src_ip_str));
            inet_ntop(AF_INET, &addr->ipv4->daddr, dst_ip_str, sizeof(dst_ip_str));
            cJSON_AddStringToObject(json_obj, "s_ip", src_ip_str);
            cJSON_AddStringToObject(json_obj, "d_ip", dst_ip_str);
            cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->ipv4->source));
            cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->ipv4->dest));
            break;
        case ADDR_TYPE_IPV6:
        case __ADDR_TYPE_IP_PAIR_V6:
            cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
            inet_ntop(AF_INET6, addr->ipv6->saddr, src_ip_str, sizeof(src_ip_str));
            inet_ntop(AF_INET6, addr->ipv6->daddr, dst_ip_str, sizeof(dst_ip_str));
            cJSON_AddStringToObject(json_obj, "s_ip", src_ip_str);
            cJSON_AddStringToObject(json_obj, "d_ip", dst_ip_str);
            cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->ipv6->source));
            cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->ipv6->dest));
            break;
        case ADDR_TYPE_VLAN:
        case ADDR_TYPE_GRE:
        case ADDR_TYPE_MPLS:
        case ADDR_TYPE_PPPOE_SES:
        case ADDR_TYPE_L2TP:
        case ADDR_TYPE_PPP:
            cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
            cJSON_AddStringToObject(json_obj, "s_ip", null_addr);
            cJSON_AddStringToObject(json_obj, "d_ip", null_addr);
            cJSON_AddNumberToObject(json_obj, "s_port", null_port);
            cJSON_AddNumberToObject(json_obj, "d_port", null_port);
            break;
        case ADDR_TYPE_PPTP:
            cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
            cJSON_AddStringToObject(json_obj, "s_ip", null_addr);
            cJSON_AddStringToObject(json_obj, "d_ip", null_addr);
            cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->pptp->C2S_call_id));
            cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->pptp->S2C_call_id));
            break;
        default:
            break;
    }

    addr_proto = layer_addr_prefix_ntop(a_stream);
    cJSON_AddStringToObject(json_obj, "trans_proto", addr_proto);

    MESA_get_stream_opt(a_stream, MSO_STREAM_TUNNEL_TYPE, &tunnel_type, &tunnel_type_size);
    if (tunnel_type == STREAM_TUNNLE_NON)
    {
        layer_addr_ntop_r(a_stream, nest_addr_buf, sizeof(nest_addr_buf));
    }
    else
    {
        stream_addr_list_ntop(a_stream, nest_addr_buf, sizeof(nest_addr_buf));
    }

    cJSON_AddStringToObject(json_obj, "addr_list", nest_addr_buf);

    return 0;
}

extern long long g_CurrentTime;
static void feedback_url_to_kafka(streaminfo *a_tcp, int http_seq, store_hash_data_t *store_hash_data)
{
	if(store_hash_data == NULL)
		return;

	char store_key[MAX_BUF_LEN];
	memset(store_key, 0, sizeof(store_key));
	int ret = append_http_query_key(a_tcp, http_seq, store_key, sizeof(store_key));
	if(ret != 0)
		return;
	cJSON *value_node = cJSON_CreateObject();
	cJSON *feedback_node = cJSON_CreateObject();	

	cJSON_AddStringToObject(value_node, "url", store_hash_data->url_buf);
    cJSON_AddStringToObject(feedback_node, "url", store_hash_data->url_buf);


    ntc_addStreamInfo_to_jsonObj(feedback_node, a_tcp);
	cJSON_AddNumberToObject(feedback_node, "found_time", store_hash_data->store_time);
    char *domain = (char *)calloc(sizeof(char), store_hash_data->url_len+1);
    form_url_to_domain(store_hash_data->url_buf, domain);
	cJSON_AddStringToObject(feedback_node, "domain", domain);

	if(store_hash_data->refer_buf != NULL)
	{
		cJSON_AddStringToObject(value_node, "referer", store_hash_data->refer_buf);
		cJSON_AddStringToObject(feedback_node, "referer", store_hash_data->refer_buf);
	}
	cJSON_AddNumberToObject(value_node, "found_time", store_hash_data->store_time);
	//char *value = cJSON_Print(value_node);
	char *value = cJSON_PrintUnformatted(value_node);

	cJSON_AddStringToObject(feedback_node, "k", store_key);
	cJSON_AddStringToObject(feedback_node, "v", value);
    cJSON_AddStringToObject(feedback_node, "cap_ip", g_item.local_ip_str);


	//char *feedback = cJSON_Print(feedback_node);
	char *feedback = cJSON_PrintUnformatted(feedback_node);

	rd_kafka_produce(g_ntc_http_collect_kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, feedback, strlen(feedback), NULL, 0, NULL);

    free(domain);
    domain=NULL;

	if(feedback != NULL)
		free(feedback);
	if(value != NULL)
		free(value);
	
	cJSON_Delete(value_node);
	cJSON_Delete(feedback_node);

	return ;
}


static void process_http_header_data(store_hash_data_t *store_hash_data, int http_seq, struct streaminfo *a_tcp)
{
	if(g_item.enable_filter ==1 && check_url_filter((char *)(store_hash_data->url_buf), store_hash_data->url_len) == 1)
	{
		if(g_item.field_handle != NULL)FS_operate(g_item.field_handle, g_item.fs_filter_num_id,0, FS_OP_ADD, 1);
	}
	else
	{
			feedback_url_to_kafka(a_tcp, http_seq, store_hash_data);
	}
	store_data_free(store_hash_data);
	store_hash_data = NULL;	
	return;
}


static int http_url_discovery_init_hash(int thread_num)
{
	if(thread_num <= 0)
		return -1;

	MESA_htable_create_args_t url_filter_hash_args;

	memset(&url_filter_hash_args, 0 ,sizeof(url_filter_hash_args));
	

	if(g_item.enable_filter == 1)
	{
		url_filter_hash_args.thread_safe = 0;
		url_filter_hash_args.hash_slot_size = 128;
		url_filter_hash_args.max_elem_num = 0;
		url_filter_hash_args.eliminate_type = HASH_ELIMINATE_ALGO_FIFO;
		url_filter_hash_args.expire_time = 0;
		url_filter_hash_args.key_comp = NULL;
		url_filter_hash_args.key2index = NULL;
		url_filter_hash_args.data_free = free;
		url_filter_hash_args.data_expire_with_condition = NULL;
		g_url_filter_hash_handle = MESA_htable_create(&url_filter_hash_args, sizeof(url_filter_hash_args));

		if(g_url_filter_hash_handle  == NULL)
		{
			MESA_htable_destroy(g_url_filter_hash_handle, NULL);
			return -1;
		}
	}
	return 0;
}

static void http_url_discovery_deinit_hash(int thread_num)
{
	if(thread_num <= 0)
		return ;


	if(g_url_filter_hash_handle  != NULL)
	{
		MESA_htable_destroy(g_url_filter_hash_handle, NULL);
	}

	return;
}

static void http_url_discovery_load_filter(const char *filename)
{
	char *line = NULL;
	char buf[MAX_BUF_LEN];
	FILE *fp = fopen(filename, "r");

	if(fp == NULL)
		return;

	while(!feof(fp))
	{
		memset(buf, 0, sizeof(MAX_BUF_LEN));
		if(NULL != fgets(buf, sizeof(buf), fp))
		{
			line = (char *)calloc(strlen(buf)-1, 1);
			memcpy(line, buf, strlen(buf)-1);
			MESA_htable_add(g_url_filter_hash_handle, (uchar *)line, strlen(buf)-1, line);
		}
	}

	fclose(fp);
	return;
}

typedef enum
{
    SEND_NONE = 0,
    SEND_KAFKA = 1,
    INDIE_KAFKA = 2,
    SEND_LOG = 4
} NTC_SEND_MODE_MASK;

static void NTC_HTTP_COLLECT_load_profile()
{
	memcpy(g_item.filter_file_path, FILTER_PATH, strlen(FILTER_PATH));
	memcpy(g_item.profile_path, PROFILE_PATH, strlen(PROFILE_PATH));

	char nic_name[64];
	MESA_load_profile_string_def(PROFILE_PATH,"SYSTEM", "NIC_NAME",nic_name,sizeof(nic_name),"eth0");
	g_item.local_ip_nr=get_ip_by_eth_name(nic_name);
	if(g_item.local_ip_nr==INADDR_NONE)
	{
		printf("get_ip_by_eth_name in %s return NULL, exit!\n", nic_name);
		exit(0);
	}

	inet_ntop(AF_INET,&(g_item.local_ip_nr),g_item.local_ip_str,sizeof(g_item.local_ip_str));
	MESA_load_profile_string_def(g_item.profile_path, PLUGIN_NAME, "log_path", g_item.log_path, sizeof(g_item.log_path), "./avlog/http_url_discovery_log");
	MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "log_level", &g_item.log_level, 30);

	MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "hash_size", &g_item.hash_size, 1024 * 1024*16);
	MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "expire_time", &g_item.expire_time, 30 * 60);

	MESA_load_profile_int_def(g_item.profile_path, PLUGIN_NAME, "reorder_num", &g_item.reorder_num, 0);

	MESA_load_profile_int_def(g_item.profile_path,PLUGIN_NAME, "kafka_mode", &g_item.kafka_mode, INDIE_KAFKA);


	MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "enable_filter", &g_item.enable_filter, 0);
	MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "enable_double", &g_item.enable_double, 0);
	MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "enable_lostlen", &g_item.enable_lostlen, 0);
	MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "enable_stat", &g_item.enable_stat, 0);
	if(g_item.enable_stat == 1)
	{	
		short fs_server_port = 0;
		char fs_server_ip[128] = "";
		MESA_load_profile_uint_def(g_item.profile_path, PLUGIN_NAME, "stat_interval", &g_item.stat_interval, 1);
		MESA_load_profile_string_def(g_item.profile_path, PLUGIN_NAME, "stat_path", g_item.stat_path, sizeof(g_item.stat_path), "./avlog/url_discovery.stat");
		MESA_load_profile_string_nodef(g_item.profile_path,PLUGIN_NAME,"stat_server_ip",fs_server_ip, sizeof(fs_server_ip));
		MESA_load_profile_short_nodef(g_item.profile_path, PLUGIN_NAME,"stat_server_port", &(fs_server_port));

		FILE *fp = fopen(g_item.stat_path, "w");
		if(fp  != NULL)
		{
			fclose(fp);
			g_item.field_handle = FS_create_handle();
			FS_set_para(g_item.field_handle, OUTPUT_DEVICE, (const void *)g_item.stat_path, strlen(g_item.stat_path));
			FS_set_para(g_item.field_handle, APP_NAME, "ntc_http", strlen("ntc_http")+1);
			int value = 2;
		        FS_set_para(g_item.field_handle, PRINT_MODE, (const void *)&value, sizeof(int));
		        value = 1;
		        FS_set_para(g_item.field_handle, PRINT_TRIGGER, (const void *)&value, sizeof(int));
		        value = 1;
		        FS_set_para(g_item.field_handle, FLUSH_BY_DATE, (const void *)&value, sizeof(int));
				value = g_item.stat_interval;
		        FS_set_para(g_item.field_handle, STAT_CYCLE, (const void *)&value, sizeof(value));
		        value=1;
		        FS_set_para(g_item.field_handle, CREATE_THREAD, &value, sizeof(value));	
				if(fs_server_port > 0 && strlen(fs_server_ip) > 0)
				{
					FS_set_para(g_item.field_handle, STATS_SERVER_IP,fs_server_ip, strlen(fs_server_ip)+1);
					FS_set_para(g_item.field_handle, STATS_SERVER_PORT,&(fs_server_port), sizeof(fs_server_port));
				}
		}
		if(g_item.field_handle != NULL)
		{
			g_item.fs_url_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "URL_NUM");
			g_item.fs_url_len_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "URL_LEN");
			g_item.fs_refer_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "REFER_NUM");
			g_item.fs_refer_len_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "REFER_LEN");
			g_item.fs_single_url_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "C2S_HTTP");
			g_item.fs_double_url_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "DOUBLE_HTTP");
			g_item.fs_s2c_http_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "S2C_HTTP");

			g_item.fs_c2s_nourl_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "C2S_NOURL");			
			g_item.fs_double_nourl_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "DOUBLE_NOURL");			
			g_item.fs_lost_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "LOST");			
			g_item.fs_err_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "ERR");
			g_item.fs_drop_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "DROP");			
			g_item.fs_filter_num_id = FS_register(g_item.field_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "FILTER");
			FS_start(g_item.field_handle);
		}		
	}
}


int NTC_HTTP_COLLECT_kaka_init(int kafka_mode)
{
	char kafka_errstr[1024];	
	MESA_load_profile_string_def(g_item.profile_path, PLUGIN_NAME, "kafka_topic", g_item.topic_buf, sizeof(g_item.topic_buf), "AIM");
	if((kafka_mode&INDIE_KAFKA)  == INDIE_KAFKA)
	{
		if(0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_brokelist", g_item.kafka_brokelist, sizeof(g_item.kafka_brokelist)))
		{
			return -1;
		}
		rd_kafka_conf_t *rdkafka_conf = rd_kafka_conf_new();
		rd_kafka_conf_set(rdkafka_conf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr));
		rd_kafka_conf_set(rdkafka_conf, "topic.metadata.refresh.interval.ms", "600000",kafka_errstr, sizeof(kafka_errstr));
		rd_kafka_conf_set(rdkafka_conf, "security.protocol", "MG", kafka_errstr, sizeof(kafka_errstr));

		if (!(g_ntc_http_collect_kafka_handle = rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr))))
		{
			return -1;
		}

		if (rd_kafka_brokers_add(g_ntc_http_collect_kafka_handle, g_item.kafka_brokelist) == 0)
		{
			return -1;
		}
	}
	else
	{
		if(0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_handle_provide_path", g_item.kafka_handle_provide_path, sizeof(g_item.kafka_handle_provide_path)))
		{
			return -1;
		}
		void * dl_handle = dlopen(g_item.kafka_handle_provide_path, RTLD_NOW|RTLD_GLOBAL);
		if(dl_handle == NULL)
		{
			return -1;
		}
		if(0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_handle_name", g_item.kafka_handle_name, sizeof(g_item.kafka_handle_name)))
		{
			return -1;
		}
		void *dl_return = dlsym(dl_handle, g_item.kafka_handle_name);

		if(dl_return == NULL)
		{
			return -1;
		}
		g_ntc_http_collect_kafka_handle = *(rd_kafka_t **)dl_return;
		if(g_ntc_http_collect_kafka_handle == NULL)
		{
			return -1;
		}        

	}
	rd_kafka_topic_conf_t*topic_conf = rd_kafka_topic_conf_new();
	g_ntc_http_collect_kafka_topic = rd_kafka_topic_new(g_ntc_http_collect_kafka_handle, g_item.topic_buf, topic_conf);
	return 0;
}

extern "C" int NTC_HTTP_COLLECT_INIT(void)
{
	int rec = 0;
	memset(&g_item, 0, sizeof(g_item));
	NTC_HTTP_COLLECT_load_profile();
	rec = http_url_discovery_init_hash(g_iThreadNum);
	if(rec != 0)
		return -1;
    rec = NTC_HTTP_COLLECT_kaka_init(g_item.kafka_mode);
	if(rec != 0)return -1;


	if(g_item.enable_filter == 1)http_url_discovery_load_filter(g_item.filter_file_path);
	g_item.log_handle = MESA_create_runtime_log_handle(g_item.log_path, g_item.log_level);

	if(g_item.log_handle == NULL)
		return -1;
	return 0;
}

extern "C" UCHAR NTC_HTTP_COLLECT_ENTRY(stSessionInfo *session_info, void **param,
		int thread_seq, struct streaminfo *a_tcp, void *a_packet)
{
	http_infor *http_info = NULL;
	if(a_tcp->dir == DIR_S2C)
	{
		FS_operate(g_item.field_handle, g_item.fs_s2c_http_num_id,0, FS_OP_ADD, 1);
		return PROT_STATE_DROPME;
	}
	if(a_tcp->dir == DIR_DOUBLE && g_item.enable_double == 0)
		return PROT_STATE_DROPME;
	if(session_info->prot_flag == HTTP_REFERER || session_info->prot_flag == HTTP_MESSAGE_URL)
	{
		if(session_info->prot_flag == HTTP_MESSAGE_URL && a_tcp->ptcpdetail->lostlen > 0 && g_item.enable_lostlen == 1)
		{
			if(g_item.field_handle != NULL)FS_operate(g_item.field_handle, g_item.fs_lost_num_id,0, FS_OP_ADD, 1);            
			if(*param != NULL)
			{
				store_data_free(*param);
				*param = NULL;
			}
			return PROT_STATE_DROPME;
		}
		if(*param == NULL)
		{
			*param = calloc(sizeof(store_hash_data_t), 1);
		}
		store_hash_data_t *p_hash_data = (store_hash_data_t *)*param;
		p_hash_data->store_time = g_CurrentTime;
		if(session_info->prot_flag == HTTP_MESSAGE_URL && session_info->buf != NULL && session_info->buflen > 0)
		{
			FS_operate(g_item.field_handle, g_item.fs_url_num_id,0, FS_OP_ADD, 1);
			FS_operate(g_item.field_handle, g_item.fs_url_len_id,0, FS_OP_ADD, session_info->buflen);

			if(a_tcp->dir == DIR_DOUBLE && g_item.field_handle != NULL)
				FS_operate(g_item.field_handle, g_item.fs_double_url_num_id,0, FS_OP_ADD, 1);
			if(a_tcp->dir == DIR_C2S && g_item.field_handle != NULL)
				FS_operate(g_item.field_handle, g_item.fs_single_url_num_id,0, FS_OP_ADD, 1);

			p_hash_data->url_buf = (char *)calloc(session_info->buflen+1, 1);
			memcpy(p_hash_data->url_buf, session_info->buf, session_info->buflen);
			p_hash_data->url_len = session_info->buflen;
		}
		if(session_info->prot_flag == HTTP_REFERER && session_info->buf != NULL && session_info->buflen > 0)
		{
			p_hash_data->refer_buf = (char *)calloc(session_info->buflen+1, 1);
			memcpy(p_hash_data->refer_buf, session_info->buf, session_info->buflen);
			p_hash_data->refer_len = session_info->buflen;

			FS_operate(g_item.field_handle, g_item.fs_refer_num_id,0, FS_OP_ADD, 1);
			FS_operate(g_item.field_handle, g_item.fs_refer_len_id,0, FS_OP_ADD, session_info->buflen);
		}
	}
	else if(session_info->prot_flag == HTTP_STATE)
	{
		if(session_info->app_info == NULL)
		{
			if(g_item.field_handle != NULL)FS_operate(g_item.field_handle, g_item.fs_err_num_id,0, FS_OP_ADD, 1);			
			return PROT_STATE_DROPME;
		}
		http_infor *a_http = (http_infor *)session_info->app_info;
		if(g_item.reorder_num > 0 && a_tcp->dir == DIR_C2S)
		{
			tcp_set_single_stream_max_unorder(a_tcp, a_tcp->dir, g_item.reorder_num);
		}		
		if((a_http->http_state == HTTP_DATA_BEGIN || a_http->http_state == HTTP_DATA_END))
		{
			if( *param!= NULL)
			{
				http_infor *p_http = (http_infor *)(session_info->app_info);
				store_hash_data_t *p_hash_data = (store_hash_data_t *)*param;
				if(p_hash_data->url_buf != NULL && p_hash_data->url_len > 0)
					process_http_header_data(p_hash_data, p_http->http_session_seq, a_tcp);
				else
				{
					if(p_http->p_url != NULL && p_http->url_len > 0)
					{

						p_hash_data->url_buf = (char *)calloc(p_http->url_len+1, 1);
						memcpy(p_hash_data->url_buf, p_http->p_url, p_http->url_len);
						p_hash_data->url_len = p_http->url_len;
						process_http_header_data(p_hash_data, p_http->http_session_seq, a_tcp);
					}
					else
					{
						if(g_item.field_handle != NULL)FS_operate(g_item.field_handle, g_item.fs_drop_num_id,0, FS_OP_ADD, 1);            
						store_data_free(*param);
						*param = NULL;
					}
				}
			}
			else
			{
				http_info = (http_infor *)session_info->app_info;
				if(http_info->method != HTTP_METHOD_CONNECT)
				{
					//if(g_item.field_handle != NULL)FS_operate(g_item.field_handle, g_item.fs_err_num_id,0, FS_OP_ADD, 1);                            
					int tcp_seq = 0;
					int tcp_seq_len = sizeof(tcp_seq);
					MESA_get_stream_opt(a_tcp, MSO_TCP_ISN_C2S, &tcp_seq, &tcp_seq_len);
					http_infor *p_http = (http_infor *)(session_info->app_info);
					char tuple_buf[MAX_BUF_LEN];
					if(a_tcp->dir == DIR_C2S)
					{
						FS_operate(g_item.field_handle, g_item.fs_c2s_nourl_num_id,0, FS_OP_ADD, 1);
					}
					if(a_tcp->dir == DIR_DOUBLE)
					{
						FS_operate(g_item.field_handle, g_item.fs_double_nourl_num_id,0, FS_OP_ADD, 1);
					}

					FS_operate(g_item.field_handle, g_item.fs_refer_num_id,0, FS_OP_ADD, 1);
					sprintf(tuple_buf, "%s#%d#0#%u", printaddr(&a_tcp->addr, a_tcp->threadnum), p_http->http_session_seq, (unsigned int)tcp_seq);	
					MESA_handle_runtime_log(g_item.log_handle, RLOG_LV_FATAL, __FUNCTION__, "C2S or DOUBLE HTTP session without URL, dir = %d, tuple5 = %s", a_tcp->dir, tuple_buf);
				}

			}
			return PROT_STATE_DROPME;
		}
	}
	return PROT_STATE_GIVEME;
}

extern "C" void NTC_HTTP_COLLECT_DESTROY(void)
{
	MESA_destroy_runtime_log_handle(g_item.log_handle);
	http_url_discovery_deinit_hash(g_iThreadNum);
	return ;
}