#include <stdio.h>
#include <string.h>
#include <sys/socket.h>//inet_addr
#include <netinet/in.h>//inet_addr
#include <arpa/inet.h>//inet_addr
#include <net/if.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <assert.h>

#include <time.h>

#include <arpa/inet.h>
#include <MESA/MESA_prof_load.h>
#include <MESA/MESA_handle_logger.h>
#include <MESA/stream.h>

#include <MESA/Maat_rule.h>
#include <MESA/field_stat2.h>
#include <MESA/cJSON.h>

#include <MESA/radius.h>
#include <librdkafka/rdkafka.h>

#include "ntc_radius_plug.h"

#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


static int NTC_RADIUS_PLUG_VERSION_20191211=0;
const char *config_file="conf/radius/radius.conf";
g_ntc_radius_plug_t g_ntc_radius_plug;
	
void set_one_maat_rule(Maat_rule_t *maat_rule)
{
	maat_rule->action=1;
	maat_rule->config_id=0;//0
	maat_rule->do_blacklist=0;
	maat_rule->do_log=1;
	maat_rule->service_id=0xA2;
}

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;
}


int get_radius_object_element(cJSON *radius_info_object, cJSON *radius_debug, radius_header_t *radius_header, radius_body_t *radius_body, int *account_index)
{
	int i=0;
		
	cJSON_AddNumberToObject(radius_debug, "radius_packet_type", radius_header->code);
	cJSON_AddNumberToObject(radius_info_object, "radius_packet_type", radius_header->code);

	for(i=0; i<radius_body->attribute_num; i++)
	{
		switch(radius_body->attribute[i].type)
		{
			case	USER_NAME:
				cJSON_AddStringToObject(radius_debug, "radius_account", (char *)(radius_body->attribute[i].value));				
				cJSON_AddStringToObject(radius_info_object, "radius_account", (char *)(radius_body->attribute[i].value));
				*account_index=i;
				break;
			case	NAS_IP_ADDRESS:
				cJSON_AddStringToObject(radius_debug, "radius_nas_ip", (char *)(radius_body->attribute[i].value));
				cJSON_AddStringToObject(radius_info_object, "radius_nas_ip", (char *)(radius_body->attribute[i].value));
				break;
			case	NAS_PORT:
				cJSON_AddNumberToObject(radius_debug, "radius_nas_port", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_nas_port", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	SERVICE_TYPE:
				cJSON_AddNumberToObject(radius_debug, "radius_service_type", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_service_type", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	FRAMED_PROTOCOL:
				cJSON_AddNumberToObject(radius_debug, "radius_framed_protocol", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_framed_protocol", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	FRAMED_IP_ADDRESS:
				cJSON_AddStringToObject(radius_debug, "radius_framed_ip", (char *)(radius_body->attribute[i].value));				
				cJSON_AddStringToObject(radius_info_object, "radius_framed_ip", (char *)(radius_body->attribute[i].value));
				break;
			case	FRAMED_IP_NETMASK:
				cJSON_AddStringToObject(radius_debug, "radius_framed_ip_netmask", (char *)(radius_body->attribute[i].value));
				break;
			case	FRAMED_ROUTING:
					cJSON_AddNumberToObject(radius_debug, "radius_framed_routing", *(unsigned int *)(radius_body->attribute[i].value));
					break;
			case	FILTER_ID:
				cJSON_AddStringToObject(radius_debug, "radius_filter_id", (char *)(radius_body->attribute[i].value));
				break;
			case	FRAMED_MTU:
					cJSON_AddNumberToObject(radius_debug, "radius_framed_mtu", *(unsigned int *)(radius_body->attribute[i].value));
					break;
			case	FRAMED_COMPRESSION:
					cJSON_AddNumberToObject(radius_debug, "radius_framed_compression", *(unsigned int *)(radius_body->attribute[i].value));
					break;
			case	CALLBACK_NUMBER:
 				cJSON_AddStringToObject(radius_debug, "radius_callback_number", (char *)(radius_body->attribute[i].value));
				cJSON_AddStringToObject(radius_info_object, "radius_callback_number", (char *)(radius_body->attribute[i].value));
				break;
			case	CALLBACK_ID:
				cJSON_AddStringToObject(radius_debug, "radius_callback_id", (char *)(radius_body->attribute[i].value));
				cJSON_AddStringToObject(radius_info_object, "radius_callback_id", (char *)(radius_body->attribute[i].value));
				break;
			case	CLASS:
					cJSON_AddStringToObject(radius_debug, "radius_class", (char *)(radius_body->attribute[i].value));
					break;
			case	SESSION_TIMEOUT:
					cJSON_AddNumberToObject(radius_debug, "radius_session_timeout", *(unsigned int *)(radius_body->attribute[i].value));
					cJSON_AddNumberToObject(radius_info_object, "radius_session_timeout", *(unsigned int *)(radius_body->attribute[i].value));
					break;
			case	IDLE_TIMEOUT:
					cJSON_AddNumberToObject(radius_debug, "radius_idle_timeout", *(unsigned int *)(radius_body->attribute[i].value));
					cJSON_AddNumberToObject(radius_info_object, "radius_idle_timeout", *(unsigned int *)(radius_body->attribute[i].value));
					break;
			case	TERMINATION_ACTION:
					cJSON_AddNumberToObject(radius_debug, "radius_termination_action", *(unsigned int *)(radius_body->attribute[i].value));
					cJSON_AddNumberToObject(radius_info_object, "radius_termination_action", *(unsigned int *)(radius_body->attribute[i].value));
					break;
			case	CALLED_STATION_ID:
 				cJSON_AddStringToObject(radius_debug, "radius_called_station_id", (char *)(radius_body->attribute[i].value));
				cJSON_AddStringToObject(radius_info_object, "radius_called_station_id", (char *)(radius_body->attribute[i].value));
				break;
			case	CALLING_STATION_ID:
  				cJSON_AddStringToObject(radius_debug, "radius_calling_station_id", (char *)(radius_body->attribute[i].value));
				cJSON_AddStringToObject(radius_info_object, "radius_calling_station_id", (char *)(radius_body->attribute[i].value));
				break;
			case	NAS_IDENTIFIER:
  				cJSON_AddStringToObject(radius_debug, "radius_nas_indentifies", (char *)(radius_body->attribute[i].value));
				break;
			case	PROXY_STATE:
				cJSON_AddNumberToObject(radius_debug, "radius_proxy_state", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_STATUS_TYPE:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_status_type", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_status_type", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_DELAY_TIME:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_delay_time", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_delay_time", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_SESSION_ID:
   				cJSON_AddStringToObject(radius_debug, "radius_acct_session_id", (char *)(radius_body->attribute[i].value));
				cJSON_AddStringToObject(radius_info_object, "radius_acct_session_id", (char *)(radius_body->attribute[i].value));
				break;
			case	ACCT_MULTI_SESSION_ID:
				cJSON_AddStringToObject(radius_debug, "radius_acct_multi_session_id", (char *)(radius_body->attribute[i].value));
				cJSON_AddStringToObject(radius_info_object, "radius_acct_multi_session_id", (char *)(radius_body->attribute[i].value));
				break;
			case	ACCT_INPUT_OCTETS:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_input_octets", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_input_octets", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_OUTPUT_OCTETS:
 				cJSON_AddNumberToObject(radius_debug, "radius_acct_output_octets", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_output_octets", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_INPUT_PACKETS:
  				cJSON_AddNumberToObject(radius_debug, "radius_acct_input_packets", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_input_packets", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_OUTPUT_PACKETS:
   				cJSON_AddNumberToObject(radius_debug, "radius_acct_output_packets", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_output_packets", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_SESSION_TIME:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_session_time", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_session_time", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_TERMINATE_CAUSE:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_terminate_cause", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_terminate_cause", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_LINK_COUNT:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_link_count", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_link_count", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_INPUT_GIGAWORDS:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_input_gigawords", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_OUTPUT_GIGAWORDS:
				cJSON_AddNumberToObject(radius_debug, "radius_acct_output_gigawords", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_EVENT_TIMESTAMP:
      			cJSON_AddNumberToObject(radius_debug, "radius_event_timestamp", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_event_timestamp", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	NAS_PORT_TYPE:
      			cJSON_AddNumberToObject(radius_debug, "radius_nas_port_type", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	ACCT_INTERIM_INTERVAL:
      			cJSON_AddNumberToObject(radius_debug, "radius_acct_interim_interval", *(unsigned int *)(radius_body->attribute[i].value));
				cJSON_AddNumberToObject(radius_info_object, "radius_acct_interim_interval", *(unsigned int *)(radius_body->attribute[i].value));
				break;
			case	NAS_PORT_ID:
				cJSON_AddStringToObject(radius_debug, "radius_nas_port_id", (char *)(radius_body->attribute[i].value));
				break;
			default:
				continue;
		}
	}

	return 0;
}

static int streamInfo2jsonObject(cJSON *json_obj, const struct streaminfo *a_stream)
{
	time_t cur_time;
	int ret=0,con_duration_ms=0;
	const char *addr_proto = NULL;
	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, "common_stream_dir", a_stream->dir);

	addr=&(a_stream->addr);
	switch(addr->addrtype)
	{
		case ADDR_TYPE_IPV4:
		case __ADDR_TYPE_IP_PAIR_V4:
			cJSON_AddNumberToObject(json_obj, "common_address_type", 4);
			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, "common_client_ip", src_ip_str);					
			cJSON_AddStringToObject(json_obj, "common_server_ip", dst_ip_str);
			cJSON_AddNumberToObject(json_obj, "common_client_port", ntohs(addr->ipv4->source));
			cJSON_AddNumberToObject(json_obj, "common_server_port", ntohs(addr->ipv4->dest));
			break;
		case ADDR_TYPE_IPV6:
		case __ADDR_TYPE_IP_PAIR_V6:
			cJSON_AddNumberToObject(json_obj, "common_address_type", 6);
			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, "common_client_ip", src_ip_str);					
			cJSON_AddStringToObject(json_obj, "common_server_ip", dst_ip_str);
			cJSON_AddNumberToObject(json_obj, "common_client_port", ntohs(addr->ipv6->source));
			cJSON_AddNumberToObject(json_obj, "common_server_port", ntohs(addr->ipv6->dest));
			break;	
		default:
			break;
	}

	cJSON_AddNumberToObject(json_obj, "common_c2s_pkt_num", a_stream->ptcpdetail->serverpktnum);
	cJSON_AddNumberToObject(json_obj, "common_s2c_pkt_num", a_stream->ptcpdetail->clientpktnum);
	cJSON_AddNumberToObject(json_obj, "common_c2s_byte_num", a_stream->ptcpdetail->serverbytes);
	cJSON_AddNumberToObject(json_obj, "common_s2c_byte_num", a_stream->ptcpdetail->clientbytes);
	
	cur_time = time(NULL);
	if(a_stream!=NULL && a_stream->ptcpdetail!=NULL)
	{
		cJSON_AddNumberToObject(json_obj, "common_start_time", a_stream->ptcpdetail->createtime);
		cJSON_AddNumberToObject(json_obj, "common_end_time", a_stream->ptcpdetail->lastmtime);
		
		con_duration_ms=(a_stream->ptcpdetail->lastmtime-a_stream->ptcpdetail->createtime)*1000;
		cJSON_AddNumberToObject(json_obj, "common_con_duration_ms", con_duration_ms);
	}
	else
	{
		con_duration_ms=0;
		cJSON_AddNumberToObject(json_obj, "common_start_time", cur_time);
		cJSON_AddNumberToObject(json_obj, "common_end_time", cur_time);
		cJSON_AddNumberToObject(json_obj, "common_con_duration_ms", con_duration_ms);
	}

	int device_id_size=sizeof(unsigned long long);
	unsigned long long device_id=(unsigned long long)g_ntc_radius_plug.device_id;
	MESA_get_stream_opt(a_stream, MSO_GLOBAL_STREAM_ID, (void *)&device_id, &device_id_size);
	cJSON_AddNumberToObject(json_obj, "common_stream_trace_id", device_id);

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

	ret=MESA_get_stream_opt(a_stream, MSO_STREAM_TUNNEL_TYPE, &tunnel_type, &tunnel_type_size);
	assert(ret==0);
	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, "common_address_list", nest_addr_buf);

	return 0;
}


char NTC_RADIUS_PLUG_ENTRY(stSessionInfo *session_info, void **pme, int thread_seq, struct streaminfo *stream, void *a_packet)
{
	int status=0;
	int config_id=0;
	int payload_len=0,index=0;
	char *payload=NULL;
	radius_info_t *radius_info=(radius_info_t *)session_info->app_info;

	if(!((g_ntc_radius_plug.interesting_type>>(radius_info->header.code))&0x1))
	{
		return PROT_STATE_GIVEME;
	}
	

	cJSON *radius_info_object=cJSON_CreateObject();
	
	streamInfo2jsonObject(radius_info_object, stream);

	cJSON_AddNumberToObject(radius_info_object, "common_policy_id", config_id);
	cJSON_AddNumberToObject(radius_info_object, "common_service", g_ntc_radius_plug.service_id);

	cJSON_AddStringToObject(radius_info_object, "common_sled_ip", g_ntc_radius_plug.local_ip);
	cJSON_AddStringToObject(radius_info_object, "common_schema_type", "RADIUS");

	cJSON *radius_debug=cJSON_Duplicate(radius_info_object, 1);
	
	get_radius_object_element(radius_info_object, radius_debug, &radius_info->header, &radius_info->body, &index);

	payload = cJSON_PrintUnformatted(radius_info_object);
	payload_len = strlen(payload);
	
	status = rd_kafka_produce(g_ntc_radius_plug.topic_rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, payload_len,
										radius_info->body.attribute[index].value, radius_info->body.attribute[index].len, NULL); 		
	
	if(status < 0)
	{
		MESA_handle_runtime_log(g_ntc_radius_plug.logger,RLOG_LV_INFO,"SEND_LOG", "sendlog to kafka is error, status: %d, topic: %s payload: %s", status, g_ntc_radius_plug.topic_name, payload);
	}
	else
	{
		MESA_handle_runtime_log(g_ntc_radius_plug.logger,RLOG_LV_INFO,"SEND_LOG", "topic: %s %s", g_ntc_radius_plug.topic_name, payload);
	}


	free(payload);
	cJSON_Delete(radius_info_object);
	radius_info_object = NULL;

	payload = cJSON_PrintUnformatted(radius_debug);
	MESA_handle_runtime_log(g_ntc_radius_plug.logger,RLOG_LV_INFO, "RADIUS_DEBUG", "payload: %s", payload);
	free(payload);
	cJSON_Delete(radius_debug);
	radius_debug = NULL;
	
	return PROT_STATE_GIVEME;
}

int NTC_RADIUS_PLUG_INIT(void)
{
	unsigned int local_ip_nr=0;
	char nic_name[32]={0};
	char kafka_errstr[1024];
	rd_kafka_t *kafka_handle = NULL;
	rd_kafka_conf_t *rdkafka_conf = NULL;
	rd_kafka_topic_conf_t *topic_conf;

	memset(&g_ntc_radius_plug,0,sizeof(g_ntc_radius_plug_t));
	MESA_load_profile_string_def(config_file, "RADIUS_PLUG", "NIC_NAME", nic_name, sizeof(nic_name), "lo");
	MESA_load_profile_string_def(config_file, "RADIUS_PLUG","LOG_PATH",g_ntc_radius_plug.pathname, sizeof(g_ntc_radius_plug.pathname), "./log/ntc_radius_plug/ntc_radius_plug");
	MESA_load_profile_int_def(config_file, "RADIUS_PLUG", "SERVICE_ID", &g_ntc_radius_plug.service_id, 0xA2);
	MESA_load_profile_int_def(config_file, "RADIUS_PLUG","LOG_LEVEL", &g_ntc_radius_plug.level, RLOG_LV_FATAL);         
	MESA_load_profile_string_def(config_file, "RADIUS_PLUG", "COLLECT_TOPIC", g_ntc_radius_plug.topic_name, sizeof(g_ntc_radius_plug.topic_name), "RADIUS-RECORD-LOG");
	MESA_load_profile_string_def(config_file, "RADIUS_PLUG", "BROKERLIST", g_ntc_radius_plug.brokerlist, sizeof(g_ntc_radius_plug.brokerlist), "127.0.0.1:9092");
	MESA_load_profile_int_def(config_file, "RADIUS_PLUG", "DEVICE_ID", &g_ntc_radius_plug.device_id, 0);

	MESA_load_profile_int_def(config_file, "RADIUS_PLUG", "PACKET_TYPE_FLAG", &g_ntc_radius_plug.interesting_type, 0x10);

	g_ntc_radius_plug.logger = MESA_create_runtime_log_handle(g_ntc_radius_plug.pathname, g_ntc_radius_plug.level);
	if(g_ntc_radius_plug.logger == NULL)
	{
		printf("RADIUS MESA_create_runtime_log_handle failed ...\n");
		return -1;
	}

	local_ip_nr=get_ip_by_eth_name(nic_name);
	if(local_ip_nr==INADDR_NONE)
	{
		MESA_handle_runtime_log(g_ntc_radius_plug.logger, RLOG_LV_FATAL, "init_soq_sendlog","get NIC_NAME: %s error.", nic_name);
		return -1;
	}
	
	inet_ntop(AF_INET, &local_ip_nr, g_ntc_radius_plug.local_ip, sizeof(g_ntc_radius_plug.local_ip));

	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, "request.required.acks", "1", kafka_errstr, sizeof(kafka_errstr));
	
	if(!(kafka_handle=rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr))))
	{
		MESA_handle_runtime_log(g_ntc_radius_plug.logger, RLOG_LV_FATAL, "RADIUS_COLLECT", "rd_kafka_new is error");
		return -1;
	}
	
	if(rd_kafka_brokers_add(kafka_handle, g_ntc_radius_plug.brokerlist) == 0)
	{
		MESA_handle_runtime_log(g_ntc_radius_plug.logger, RLOG_LV_FATAL, "RADIUS_COLLECT", "rd_kafka_brokers_add is error, broker_list: %s", g_ntc_radius_plug.brokerlist);
		return -1;
	}
	
	topic_conf = rd_kafka_topic_conf_new();
	g_ntc_radius_plug.topic_rkt = rd_kafka_topic_new(kafka_handle, g_ntc_radius_plug.topic_name, topic_conf);
	if(g_ntc_radius_plug.topic_rkt==NULL)
	{
		MESA_handle_runtime_log(g_ntc_radius_plug.logger, RLOG_LV_FATAL, "RADIUS_COLLECT", "rd_kafka_topic_new is error, topic: %s", g_ntc_radius_plug.topic_name);
		return -1;
	}	

	printf("INIT NTC_RADIUS_PLUG SUCCESS, VERSION: %d\n", NTC_RADIUS_PLUG_VERSION_20191211);
	
	return 0;
}

void NTC_RADIUS_PLUG_DESTROY(void)
{
	
}