DNS Decoder create version

1 files changed, 1936 insertions, 0 deletions

diff --git a/src/dns_decoder.cpp b/src/dns_decoder.cpp
new file mode 100644
index 0000000..8278c8d
--- /dev/null
+++ b/src/dns_decoder.cpp
@@ -0,0 +1,1936 @@
+/*
+**********************************************************************************************
+*	File: dns_decoder.c
+*   Description:
+*	Authors: Liu XueLi <[email protected]>
+*	Date:    2024-02-07
+*   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
+***********************************************************************************************
+*/
+
+#include <stdio.h>
+#include <string.h>
+#include <assert.h>
+#include <stdlib.h>
+
+#include "dns_decoder.h"
+#include "toml/toml.h"
+#include "uthash/uthash.h"
+#include "uthash/utlist.h"
+
+#include "fieldstat/fieldstat_easy.h"
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "stellar/utils.h"
+#include "stellar/session.h"
+#include "stellar/stellar.h"
+#include "stellar/session_mq.h"
+#include "stellar/session_exdata.h"
+
+#ifdef __cplusplus
+}
+#endif
+
+#define DNS_HEADER_SIZE 12
+
+#define	DNS_DECODER_FALSE	0
+#define	DNS_DECODER_TRUE	1
+
+// INIT, SUCCESS, FAILURE
+#define	DNS_RR_STATUS_INIT		1
+#define	DNS_RR_STATUS_SUCCESS	2
+#define	DNS_RR_STATUS_FAILURE	3
+#define	DNS_RR_STATUS_CONTINUE	4
+
+#define	DNS_COMPRESSION_POINTER_00C0	0xc0
+
+#define	TAG_KEY_MESSAGE_TYPE	"message_type"
+#define	TAG_KEY_MESSAGE_STATUS	"message_status"
+#define	TAG_KEY_DECODE_FIELD	"decode_field"
+
+#define	TAG_KEY_IP_PROTOCOL	"ip_protocol"
+#define	TAG_VALUE_IP_PROTOCOL_TCP	"tcp"
+#define	TAG_VALUE_IP_PROTOCOL_UDP	"udp"
+
+#define	TAG_KEY_IP_VERSION	"ip_version"
+#define	TAG_VALUE_IP_VERSION_IPV4	"ipv4"
+#define	TAG_VALUE_IP_VERSION_IPV6	"ipv6"
+
+#define DNS_MESSAGE_MAGIC	0x53535353
+#define	DNS_DECODER_TOML_PATH "./etc/dns/dns_decoder.toml"
+
+struct dns_message 
+{
+	int32_t magic;
+    enum dns_message_type type;
+
+	int32_t current_trans_idx;
+	uint8_t decode_rr_status;
+	uint16_t trans_identifier_id;
+	struct dns_flag flag;
+	uint16_t n_question;
+	
+    uint16_t n_answer_rr;
+    uint16_t n_authority_rr;
+    uint16_t n_additional_rr;
+	uint16_t n_real_answer_rr;
+    uint16_t n_real_authority_rr;
+    uint16_t n_real_additional_rr;
+	struct dns_query_question question;
+	struct dns_resource_record *answer_rr;
+    struct dns_resource_record *authority_rr;
+    struct dns_resource_record *additional_rr;
+
+	int32_t rr_capacity_offset;
+	int32_t rr_capacity_sz;
+	struct dns_resource_record *rr_capacity;
+
+	uint8_t *payload;
+	size_t payload_sz;
+	size_t payload_offset;
+};
+
+struct dns_decoder_stat
+{
+    int *metric_id;
+	int per_thread_enable;
+	int interval_second;
+	char name[DNS_NAME_MAX];
+	char path[DNS_NAME_MAX];
+    struct fieldstat_easy *fse;
+};
+
+struct message_schema
+{
+    int32_t sub_id;
+    int32_t topic_id;
+    const char *topic_name;
+    session_msg_free_cb_func *free_cb;
+    on_session_msg_cb_func *on_cb;
+};
+
+struct dns_decoder_plugin_env 
+{
+    int32_t plugin_id;
+	struct stellar *st;
+	uint16_t *net_port;
+	int32_t n_net_port;
+	int32_t max_rr_num;
+	int32_t max_cache_trans_num;
+    struct message_schema dns;
+	struct message_schema udp;
+	struct message_schema tcp_stream;
+	struct dns_message *per_thread_data_msg;
+	// struct dns_message *per_thread_trans_new;
+	// struct dns_message *per_thread_trans_free;
+
+	struct dns_decoder_stat stat;
+};
+
+struct dns_transaction
+{
+	int32_t message_id;
+	int32_t trans_idx;
+	UT_hash_handle hh;
+	struct dns_transaction *prev;
+	struct dns_transaction *next;
+};
+
+struct dns_decoder_context
+{
+    int32_t trans_count;
+    size_t cache_sz;
+	uint8_t cache_ptr[2];
+
+	short trans_list_num;
+	struct dns_transaction *trans_list_head;
+	struct dns_transaction *trans_hash;
+};
+
+struct dns_header 
+{
+    uint16_t id;
+#if __BYTE_ORDER==__LITTLE_ENDIAN
+	uint8_t rd:1;
+	uint8_t tc:1;
+	uint8_t aa:1;
+	uint8_t opcode:4;
+	uint8_t qr:1;
+	uint8_t rcode:4;
+	uint8_t z:3;
+	uint8_t ra:1;
+#elif __BYTE_ORDER==__BIG_ENDIAN
+	uint8_t qr:1;
+	uint8_t opcode:4;
+	uint8_t aa:1;
+	uint8_t tc:1;
+	uint8_t rd:1;
+	uint8_t ra:1;
+	uint8_t z:3;
+	uint8_t rcode:4;
+#endif
+	uint16_t qdcount;
+	uint16_t ancount;
+	uint16_t aucount; //authority count
+	uint16_t adcount; //additional count
+};
+
+enum LOCAL_STAT_COUNTER
+{
+	LOCAL_STAT_COUNTER_UNKNOWN=0,
+	LOCAL_STAT_COUNTER_SESSION,
+	LOCAL_STAT_COUNTER_PACKETS,
+	LOCAL_STAT_COUNTER_BYTES,
+	LOCAL_STAT_COUNTER_SEND,
+	LOCAL_STAT_COUNTER_RECV,
+	LOCAL_STAT_COUNTER_NEW,
+	LOCAL_STAT_COUNTER_FREE,
+	LOCAL_STAT_COUNTER_OK,
+	LOCAL_STAT_COUNTER_ERROR,
+	LOCAL_STAT_COUNTER_MAX
+};
+
+int32_t dns_read_u8(uint8_t *payload, size_t payload_sz, size_t *payload_offset, uint8_t *value)
+{
+	if(payload_sz<(*payload_offset)+1)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	if(value!=NULL)
+	{
+		*value=(uint8_t)payload[(*payload_offset)];
+	}
+	
+	(*payload_offset)++;
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_read_be_u16(uint8_t *payload, size_t payload_sz, size_t *payload_offset, uint16_t *value)
+{
+	if(payload_sz<(*payload_offset)+2)
+	{
+		return DNS_DECODER_FALSE;
+	}
+	
+	if(value!=NULL)
+	{
+		*value=ntohs(*(uint16_t *)(payload+(*payload_offset)));
+	}
+
+	(*payload_offset)+=2;
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_read_be_u24(uint8_t *payload, size_t payload_sz, size_t *payload_offset, uint8_t *value)
+{
+	if(payload_sz<(*payload_offset)+3)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	if(value!=NULL)
+	{
+		dns_read_u8(payload, payload_sz, payload_offset, &value[2]);
+		dns_read_u8(payload, payload_sz, payload_offset, &value[1]);
+		dns_read_u8(payload, payload_sz, payload_offset, &value[0]);
+	}
+	else
+	{
+		(*payload_offset)+=3;
+	}
+	
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_read_be_u32(uint8_t *payload, size_t payload_sz, size_t *payload_offset, uint32_t *value)
+{
+	if(payload_sz<(*payload_offset)+4)
+	{	
+		return DNS_DECODER_FALSE;
+	}
+	
+	if(value!=NULL)
+	{
+		*value=ntohl(*(uint32_t *)(payload+(*payload_offset)));
+	}
+	
+	(*payload_offset)+=4;
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_read_variable_dstring(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct dstring *dstr, size_t dcopy_sz)
+{
+	if(payload_sz<(*payload_offset)+dcopy_sz)
+	{
+		dstr->value_sz=0;
+		dstr->value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	size_t value_sz=MIN(dcopy_sz, sizeof(dstr->value) - dstr->value_sz - 1);
+	if(value_sz>0) 
+	{
+		memcpy(dstr->value+dstr->value_sz, payload+(*payload_offset), value_sz);
+		dstr->value_sz+=value_sz;
+	}
+
+	dstr->value[dstr->value_sz]='\0';
+	
+	(*payload_offset)+=dcopy_sz;
+
+	return DNS_DECODER_TRUE;
+}
+
+void dns_flag_copy(struct dns_flag *flag, struct dns_header *dns_hdr)
+{
+    flag->qr=dns_hdr->qr;
+    flag->opcode=dns_hdr->opcode;
+    flag->aa=dns_hdr->aa;
+    flag->tc=dns_hdr->tc;
+    flag->rd=dns_hdr->rd;
+    flag->ra=dns_hdr->ra;
+    flag->z=dns_hdr->z;
+    flag->rcode=dns_hdr->rcode;
+}
+
+void populate_dns_header(struct dns_header *dns_hdr, const uint8_t *payload)
+{
+	struct dns_header *tmp=(struct dns_header *)payload;
+
+	dns_hdr->qr=tmp->qr;
+	dns_hdr->opcode= tmp->opcode;
+	dns_hdr->aa=tmp->aa;
+	dns_hdr->tc=tmp->tc;
+	dns_hdr->rd=tmp->rd;
+	dns_hdr->ra=tmp->ra;
+	dns_hdr->z=tmp->z;
+	dns_hdr->rcode=tmp->rcode;
+
+	dns_hdr->id=ntohs(tmp->id);
+	dns_hdr->qdcount=ntohs(tmp->qdcount);
+	dns_hdr->ancount=ntohs(tmp->ancount);
+	dns_hdr->aucount=ntohs(tmp->aucount);
+	dns_hdr->adcount=ntohs(tmp->adcount);
+}
+
+int32_t validate_dns_header(struct dns_header *dns_hdr)
+{ 
+	if(dns_hdr->qdcount>1)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_dstring_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct dstring *str)
+{
+	uint8_t str_sz=0;
+	int32_t ret=dns_read_u8(payload, payload_sz, payload_offset, &str_sz);
+	if(ret==DNS_DECODER_FALSE)
+	{
+		str->value_sz=0;
+		str->value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	ret=dns_read_variable_dstring(payload, payload_sz, payload_offset, str, str_sz);
+	if(ret==DNS_DECODER_FALSE)
+	{
+		str->value_sz=0;
+		str->value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+
+static size_t integer_tag_fill(struct fieldstat_tag *tag, const char *key, long long value)
+{
+    if(tag==NULL || key==NULL || strlen(key)==0)
+    {
+        return 0;
+    }
+
+    tag->type=TAG_INTEGER;
+    tag->key=key;
+    tag->value_longlong=value;
+    return 1;
+}
+
+static size_t string_tag_fill(struct fieldstat_tag *tag, const char *key, const char *value)
+{
+    if(tag==NULL || key==NULL || value==NULL || strlen(key)==0 || strlen(value)==0)
+    {
+        return 0;
+    }
+
+    tag->type=TAG_CSTRING;
+    tag->key=key;
+    tag->value_str=value;
+    return 1;
+}
+
+void dns_decoder_local_file_counter_incby(struct dns_decoder_plugin_env *plugin_env, enum LOCAL_STAT_COUNTER idx, const char *tag_key[], const char *tag_value[], size_t n_tags, long long increment, int thread_id)
+{
+    if(plugin_env->stat.fse==NULL || thread_id < 0)
+    {
+        return ;
+    }
+    
+    size_t tags_offset=0;
+    struct fieldstat_tag tags[n_tags+2]={0};
+	
+	tags_offset+=string_tag_fill(&(tags[tags_offset]), "decoder", "dns");
+	
+	if(plugin_env->stat.per_thread_enable==DNS_DECODER_TRUE)
+	{
+		tags_offset+=integer_tag_fill(&(tags[tags_offset]), "thread", thread_id);
+	}
+	
+    for(size_t i=0; i<n_tags; i++)
+    {
+        tags_offset+=string_tag_fill(&(tags[tags_offset]), tag_key[i], tag_value[i]);
+    }
+
+    fieldstat_easy_counter_incrby(plugin_env->stat.fse, thread_id, plugin_env->stat.metric_id[idx], ((tags_offset==0) ? NULL : tags), tags_offset, increment);
+}
+
+// https://dotat.at/@/2022-07-01-dns-compress.html
+// https://en.wikipedia.org/wiki/Domain_name
+int32_t dns_decompressed_domain_name(uint8_t *payload, size_t payload_sz, size_t *payload_offset, uint8_t *domain, size_t *domain_sz)
+{
+	uint8_t np=0;
+	size_t domain_offset=0;
+	size_t offset=(*payload_offset);
+	uint8_t *cur_pos=payload+offset;
+	uint8_t has_compression=DNS_DECODER_FALSE;
+
+	while(payload_sz>=offset)
+	{
+		uint8_t first_byte=cur_pos[0];
+		if(0==first_byte)
+		{
+			break;
+		}
+
+		if(DNS_COMPRESSION_POINTER_00C0==(first_byte&DNS_COMPRESSION_POINTER_00C0)) 
+		{
+			if(payload_sz < offset+1)
+			{
+				return DNS_DECODER_FALSE;
+			}
+
+			uint8_t second_byte=cur_pos[1];
+			/* udp payload offset */
+			size_t one_label_sz=((first_byte&0x03f) << 8)+second_byte;
+
+			if(has_compression==DNS_DECODER_FALSE)
+			{
+				offset+=2;
+				(*payload_offset)=offset;
+				has_compression=DNS_DECODER_TRUE;
+			}
+
+			offset=one_label_sz;
+			cur_pos=payload+one_label_sz;
+			/* udp payload offset=dns_header (12bytes) + payload */
+			if(payload_sz < one_label_sz)
+			{
+				return DNS_DECODER_FALSE;
+			}
+				
+			/* too many pointers. */
+			if (np++ > 16) 
+			{
+				return DNS_DECODER_FALSE;
+			}
+
+			continue;
+		}
+
+		size_t one_label_sz=first_byte;
+		cur_pos++;
+		offset++;
+		if(payload_sz < offset+one_label_sz)
+		{
+			return DNS_DECODER_FALSE;
+		}
+			
+		if(domain_offset+one_label_sz >= (*domain_sz-1)) 
+		{
+			return DNS_DECODER_FALSE;
+		}
+			
+		memcpy(domain+domain_offset, cur_pos, one_label_sz);
+		domain_offset+=one_label_sz;
+		domain[domain_offset++]='.';
+		offset+=one_label_sz;
+		cur_pos+=one_label_sz;
+	}
+	
+	if(has_compression==DNS_DECODER_FALSE)
+	{
+		(*payload_offset)=offset+1;
+	}
+	
+	/* omit last '.' */
+	int32_t idx=((domain_offset>0) ? (domain_offset-1) : 0);
+	domain[idx]='\0';
+	(*domain_sz)=domain_offset;
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_hinfo_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_hinfo *hinfo)
+{
+	uint8_t cpu_sz=0;
+	int32_t ret=dns_read_u8(payload, payload_sz, payload_offset, &cpu_sz);
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	ret=dns_read_variable_dstring(payload, payload_sz, payload_offset, &(hinfo->cpu), cpu_sz);
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	uint8_t os_sz=0;
+	ret=dns_read_u8(payload, payload_sz, payload_offset, &os_sz);
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	ret=dns_read_variable_dstring(payload, payload_sz, payload_offset, &(hinfo->os), os_sz);
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+	
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_minfo_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_minfo *minfo)
+{
+	minfo->rmailbx.value_sz=sizeof(minfo->rmailbx.value);
+	int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, minfo->rmailbx.value, &(minfo->rmailbx.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		minfo->rmailbx.value_sz=0;
+		minfo->rmailbx.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	minfo->emailbx.value_sz=sizeof(minfo->emailbx.value);
+	ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, minfo->emailbx.value, &(minfo->emailbx.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		minfo->emailbx.value_sz=0;
+		minfo->emailbx.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_mx_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_mx *mx, uint16_t rdata_sz)
+{
+	int32_t ret=dns_read_be_u16(payload, payload_sz, payload_offset, &(mx->preference));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	if(rdata_sz<2)
+	{
+		(*payload_offset)+=rdata_sz;
+		return DNS_DECODER_TRUE;
+	}
+
+	if(rdata_sz-2 < (payload)[*payload_offset])
+	{
+		ret=dns_read_variable_dstring(payload, payload_sz, payload_offset, &(mx->exchange), rdata_sz-2);
+		if(ret==DNS_DECODER_FALSE)
+		{
+			mx->exchange.value_sz=0;
+			mx->exchange.value[0]='\0';
+			return DNS_DECODER_FALSE;
+		}
+	}
+	else
+	{
+		mx->exchange.value_sz=sizeof(mx->exchange.value);
+		ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, mx->exchange.value, &(mx->exchange.value_sz));
+		if(ret==DNS_DECODER_FALSE)
+		{
+			mx->exchange.value_sz=0;
+			mx->exchange.value[0]='\0';
+			return DNS_DECODER_FALSE;
+		}
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_soa_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_soa *soa, uint16_t rdata_sz)
+{
+	soa->mname.value_sz=sizeof(soa->mname.value);
+	int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, soa->mname.value, &(soa->mname.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		soa->mname.value_sz=0;
+		soa->mname.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	soa->rname.value_sz=sizeof(soa->rname.value);
+	ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, soa->rname.value, &(soa->rname.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		soa->rname.value_sz=0;
+		soa->rname.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	if(payload_sz<(*payload_offset)+sizeof(uint32_t)*5)
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+	
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(soa->serial));
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(soa->refresh));
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(soa->retry));
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(soa->expire));
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(soa->minimum));
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_rp_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_rp *rp)
+{
+	rp->txt_rr.value_sz=sizeof(rp->txt_rr.value);
+	int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, rp->txt_rr.value, &(rp->txt_rr.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		rp->txt_rr.value_sz=0;
+		rp->txt_rr.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	rp->mailbox.value_sz=sizeof(rp->mailbox.value);
+	ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, rp->mailbox.value, &(rp->mailbox.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		rp->mailbox.value_sz=0;
+		rp->mailbox.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_wks_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_wks *wks, uint16_t rdata_sz)
+{
+	if((payload_sz<(*payload_offset)+5) || (rdata_sz<5))
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(wks->addr));
+	dns_read_u8(payload, payload_sz, payload_offset, &(wks->protocol));
+
+	if(rdata_sz>5)
+	{
+		wks->bitmap=payload+(*payload_offset);
+		wks->size=rdata_sz-5;
+		(*payload_offset)+=rdata_sz-5;
+	}
+	else
+	{
+		wks->size=0;
+		wks->bitmap=NULL;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_srv_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_srv *srv, uint16_t rdata_sz)
+{
+	if((payload_sz<(*payload_offset)+sizeof(uint16_t)*3) || (rdata_sz<sizeof(uint16_t)*3))
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(srv->priority));
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(srv->weight));
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(srv->port));
+		
+	srv->target.value_sz=sizeof(srv->target.value);
+	int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, srv->target.value, &(srv->target.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		srv->target.value_sz=0;
+		srv->target.value[0]='\0';
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_ds_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_ds *ds, uint16_t rdata_sz)
+{
+	if((payload_sz<(*payload_offset)+sizeof(uint16_t)+2) || (rdata_sz<sizeof(uint16_t)+2))
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(ds->key_tag));
+	dns_read_u8(payload, payload_sz, payload_offset, &(ds->algo));
+	dns_read_u8(payload, payload_sz, payload_offset, &(ds->digest_type));
+		
+	if(rdata_sz>4)
+	{
+		ds->digest=payload+(*payload_offset);
+		ds->digest_len=rdata_sz-4;
+		(*payload_offset)+=rdata_sz-4;
+	}
+	else
+	{
+		ds->digest_len=0;
+		ds->digest=NULL;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_rrsig_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_rrsig *rrsig, uint16_t rdata_sz)
+{
+	if((payload_sz<(*payload_offset)+18) || (rdata_sz<18))
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	size_t offset=(*payload_offset);
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(rrsig->type_covered));
+	dns_read_u8(payload, payload_sz, payload_offset, &(rrsig->algo));
+	dns_read_u8(payload, payload_sz, payload_offset, &(rrsig->labels));
+
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(rrsig->original_ttl));
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(rrsig->sig_expiration));
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(rrsig->sig_inception));
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(rrsig->key_tag));
+
+	if(rdata_sz>18)
+	{
+		rrsig->signer_name.value_sz=sizeof(rrsig->signer_name.value);
+		int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, rrsig->signer_name.value, &(rrsig->signer_name.value_sz));
+		if(ret==DNS_DECODER_FALSE)
+		{
+			rrsig->signer_name.value_sz=0;
+			rrsig->signer_name.value[0]='\0';
+			return DNS_DECODER_FALSE;
+		}
+	}
+	else
+	{
+		rrsig->signer_name.value_sz=0;
+		rrsig->signer_name.value[0]='\0';
+	}
+
+	offset=(*payload_offset)-offset;
+	if(rdata_sz > offset)
+	{
+		rrsig->signature=payload+(*payload_offset);
+		rrsig->signature_len=rdata_sz-offset;
+		(*payload_offset)+=rdata_sz-offset;
+	}
+	else
+	{
+		rrsig->signature_len=0;
+		rrsig->signature=NULL;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_type_bitmap_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct dstring *type_bit_maps, uint16_t type_bit_maps_sz)
+{
+	if(type_bit_maps_sz==0 || payload_sz<(*payload_offset)+type_bit_maps_sz)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	size_t cur_offset=0;
+	size_t map_offset=0;
+	uint8_t *cur_pos=payload+(*payload_offset);
+
+	type_bit_maps->value_sz=sizeof(type_bit_maps->value);
+
+	while((type_bit_maps_sz-cur_offset) > 0) 
+	{
+		if(map_offset+2 > type_bit_maps->value_sz)
+		{
+			break;
+		}
+
+		type_bit_maps->value[map_offset++]=cur_pos[cur_offset++];		
+		type_bit_maps->value[map_offset++]=cur_pos[cur_offset];
+		size_t blocksize=cur_pos[cur_offset++];	
+
+		size_t length=MIN(type_bit_maps->value_sz-map_offset, blocksize);
+		if (length==0 || blocksize > (type_bit_maps_sz - cur_offset)) 
+		{
+			break;
+		}
+		
+		memcpy(type_bit_maps->value + map_offset, cur_pos + cur_offset, length);
+		cur_offset += blocksize;
+		map_offset += length;
+	}
+
+	type_bit_maps->value_sz=map_offset;
+	(*payload_offset)+=type_bit_maps_sz;
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_nsec_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_nsec *nsec, uint16_t rdata_sz)
+{
+	size_t offset=(*payload_offset);
+	nsec->next_domain.value_sz=sizeof(nsec->next_domain.value);
+	int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, nsec->next_domain.value, &(nsec->next_domain.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		nsec->next_domain.value_sz=0;
+		nsec->next_domain.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	ret=dns_resource_record_type_bitmap_decode(payload, payload_sz, payload_offset, &(nsec->type_bit_maps), rdata_sz - ((*payload_offset)-offset));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		nsec->type_bit_maps.value_sz=0;
+		nsec->type_bit_maps.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_nsec3_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_nsec3 *nsec3, uint16_t rdata_sz)
+{
+	if(payload_sz<(*payload_offset)+6)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	size_t offset=(*payload_offset);
+
+	dns_read_u8(payload, payload_sz, payload_offset, &(nsec3->hash_algo));
+	dns_read_u8(payload, payload_sz, payload_offset, &(nsec3->flags));
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(nsec3->iteration));
+
+	int32_t ret=dns_read_u8(payload, payload_sz, payload_offset, &(nsec3->salt_len));
+	if(ret==DNS_DECODER_FALSE || (payload_sz < (*payload_offset)+nsec3->salt_len))
+	{
+		nsec3->hash_len=0;
+		nsec3->next_hash_owner=NULL;
+		return DNS_DECODER_FALSE;
+	}
+
+	nsec3->salt_value=((nsec3->salt_len > 0) ? (payload+(*payload_offset)) : NULL);
+	(*payload_offset)+=nsec3->salt_len;
+	
+	ret=dns_read_u8(payload, payload_sz, payload_offset, &(nsec3->hash_len));
+	if(ret==DNS_DECODER_FALSE || (payload_sz < (*payload_offset)+nsec3->hash_len))
+	{
+		nsec3->hash_len=0;
+		nsec3->next_hash_owner=NULL;
+		return DNS_DECODER_FALSE;
+	}
+
+	nsec3->next_hash_owner=((nsec3->hash_len > 0) ? (payload+(*payload_offset)) : NULL);
+	(*payload_offset)+=nsec3->hash_len;
+
+	ret=dns_resource_record_type_bitmap_decode(payload, payload_sz, payload_offset, &(nsec3->type_bit_maps), (uint16_t)(rdata_sz - ((*payload_offset)-offset)));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		nsec3->type_bit_maps.value_sz=0;
+		nsec3->type_bit_maps.value[0]='\0';
+		return DNS_DECODER_FALSE;
+	}
+	
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_nsec3param_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_nsec3param *nsec3param, uint16_t rdata_sz)
+{
+	if(payload_sz<(*payload_offset)+5)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	dns_read_u8(payload, payload_sz, payload_offset, &(nsec3param->hash_algo));
+	dns_read_u8(payload, payload_sz, payload_offset, &(nsec3param->flags));
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(nsec3param->iteration));
+	int32_t ret=dns_read_u8(payload, payload_sz, payload_offset, &(nsec3param->salt_len));
+	if(ret==DNS_DECODER_FALSE || (payload_sz < (*payload_offset)+nsec3param->salt_len))
+	{
+		nsec3param->salt_len=0;
+		nsec3param->salt_value=NULL;
+		return DNS_DECODER_FALSE;
+	}
+
+	nsec3param->salt_value=((nsec3param->salt_len > 0) ? (payload+(*payload_offset)) : NULL);
+	(*payload_offset)+=nsec3param->salt_len;
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_dnskey_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct rdata_dnskey *dnskey, uint16_t rdata_sz)
+{
+	if(payload_sz<(*payload_offset)+4)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(dnskey->flags));
+	dns_read_u8(payload, payload_sz, payload_offset, &(dnskey->protocol));
+	dns_read_u8(payload, payload_sz, payload_offset, &(dnskey->algo));
+
+	if(rdata_sz>4)
+	{
+		dnskey->public_key=payload+(*payload_offset);
+		dnskey->public_key_len=rdata_sz-4;
+		(*payload_offset)+=rdata_sz-4;
+	}
+	else
+	{
+		dnskey->public_key_len=0;
+		dnskey->public_key=NULL;
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_specific_field_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct dns_resource_record *rr)
+{
+	int32_t ret=DNS_DECODER_TRUE;
+	size_t offset=(*payload_offset);
+	
+	rr->rdata.string.value_sz=0;
+	memset(rr->rdata.string.value, 0, sizeof(rr->rdata.string.value));
+
+	switch(rr->type)
+	{
+		case DNS_RR_TYPE_MB:
+		case DNS_RR_TYPE_MD:
+		case DNS_RR_TYPE_MF:
+		case DNS_RR_TYPE_MG:
+		case DNS_RR_TYPE_MR:
+		case DNS_RR_TYPE_NS:
+		case DNS_RR_TYPE_PTR:
+		case DNS_RR_TYPE_DNAME:
+		case DNS_RR_TYPE_CNAME:
+			rr->rdata.string.value_sz=sizeof(rr->rdata.string.value);
+			ret=dns_decompressed_domain_name(payload, payload_sz, &offset, rr->rdata.string.value, &(rr->rdata.string.value_sz));
+			break;
+		case DNS_RR_TYPE_HINFO:
+			ret=dns_resource_record_hinfo_decode(payload, payload_sz, &offset, &(rr->rdata.hinfo));
+			break;
+		case DNS_RR_TYPE_MINFO:
+			ret=dns_resource_record_minfo_decode(payload, payload_sz, &offset, &(rr->rdata.minfo));
+			break;
+		case DNS_RR_TYPE_MX:
+			ret=dns_resource_record_mx_decode(payload, payload_sz, &offset, &(rr->rdata.mx), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_SOA:
+			ret=dns_resource_record_soa_decode(payload, payload_sz, &offset, &(rr->rdata.soa), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_A:
+			ret=dns_read_variable_dstring(payload, payload_sz, &offset, &(rr->rdata.a), sizeof(uint32_t));
+			break;
+		case DNS_RR_TYPE_AAAA:
+			ret=dns_read_variable_dstring(payload, payload_sz, &offset, &(rr->rdata.aaaa), 16);
+			break;
+		case DNS_RR_TYPE_ISDN:
+			ret=dns_read_variable_dstring(payload, payload_sz, &offset, &(rr->rdata.isdn), sizeof(uint8_t));
+			break;
+		case DNS_RR_TYPE_TXT:
+			ret=dns_dstring_decode(payload, payload_sz, &offset, &(rr->rdata.txt));
+			break;
+		case DNS_RR_TYPE_RP:
+			ret=dns_resource_record_rp_decode(payload, payload_sz, &offset, &(rr->rdata.rp));
+			break;
+		case DNS_RR_TYPE_NULL:
+			ret=dns_dstring_decode(payload, payload_sz, &offset, &(rr->rdata.null));
+			break;
+		case DNS_RR_TYPE_WKS:
+			ret=dns_resource_record_wks_decode(payload, payload_sz, &offset, &(rr->rdata.wks), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_SRV:
+			ret=dns_resource_record_srv_decode(payload, payload_sz, &offset, &(rr->rdata.srv), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_DS:
+		case DNS_RR_TYPE_DLV:
+			ret=dns_resource_record_ds_decode(payload, payload_sz, &offset, &(rr->rdata.ds), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_RRSIG:
+			ret=dns_resource_record_rrsig_decode(payload, payload_sz, &offset, &(rr->rdata.rrsig), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_NSEC:
+			ret=dns_resource_record_nsec_decode(payload, payload_sz, &offset, &(rr->rdata.nsec), rr->rdlength);		
+			break;
+			case DNS_RR_TYPE_NSEC3:
+			ret=dns_resource_record_nsec3_decode(payload, payload_sz, &offset, &(rr->rdata.nsec3), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_NSEC3PARAM:
+			ret=dns_resource_record_nsec3param_decode(payload, payload_sz, &offset, &(rr->rdata.nsec3param), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_DNSKEY:
+			ret=dns_resource_record_dnskey_decode(payload, payload_sz, &offset, &(rr->rdata.dnskey), rr->rdlength);
+			break;
+		case DNS_RR_TYPE_UNKNOWN:
+			ret=dns_dstring_decode(payload, payload_sz, &offset, &(rr->rdata.unknown));
+			break;
+		case DNS_RR_TYPE_OPT:
+			return DNS_DECODER_TRUE; // do nothing
+		default:
+			fprintf(stderr, "No support dns rr type, type: %d", rr->type);
+			break;
+	}
+
+	(*payload_offset)+=rr->rdlength;
+	
+	return ret;
+}
+
+int32_t dns_resource_record_common_field_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct dns_resource_record *rr)
+{
+	rr->qname.value_sz=sizeof(rr->qname.value);
+	int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, rr->qname.value, &(rr->qname.value_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	if(payload_sz<(*payload_offset)+sizeof(uint16_t))
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	uint16_t type=0;
+	ret=dns_read_be_u16(payload, payload_sz, payload_offset, &type);
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	rr->type=(enum dns_rr_type)type;
+	if(rr->type==DNS_CLASS_UNKNOWN)
+	{
+		return DNS_DECODER_FALSE;
+	}
+	
+	if(payload_sz< ((*payload_offset)+sizeof(uint16_t)*2+sizeof(uint32_t)))
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(rr->rr_class));
+	dns_read_be_u32(payload, payload_sz, payload_offset, &(rr->ttl));
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(rr->rdlength));
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_resource_record_decode(uint8_t *payload, size_t payload_sz, size_t *payload_offset, struct dns_resource_record *rr, uint16_t n_rr)
+{
+	if(payload_sz<(*payload_offset)+1)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	if(rr==NULL || n_rr==0)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+    for(size_t i=0; i<n_rr; i++) 
+	{
+		if(payload_sz<(*payload_offset)+1)
+		{
+			return DNS_DECODER_FALSE;
+		}
+		
+		int32_t ret=dns_resource_record_common_field_decode(payload, payload_sz, payload_offset, &rr[i]);
+		if(ret==DNS_DECODER_FALSE) 
+		{
+			return DNS_DECODER_FALSE;
+		}
+
+		ret=dns_resource_record_specific_field_decode(payload, payload_sz, payload_offset, &rr[i]);
+		if(ret==DNS_DECODER_FALSE)
+		{
+			return DNS_DECODER_FALSE;
+		}
+	}
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_query_question_decode(struct dns_query_question *question, uint8_t *payload, size_t payload_sz, size_t *payload_offset)
+{
+	if(payload_sz<(*payload_offset)+1)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	question->qname_sz=sizeof(question->qname);
+	int32_t ret=dns_decompressed_domain_name(payload, payload_sz, payload_offset, question->qname, &(question->qname_sz));
+	if(ret==DNS_DECODER_FALSE)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	if(payload_sz<(*payload_offset)+sizeof(uint16_t)*2)
+	{
+		return DNS_DECODER_FALSE;
+	}
+
+	uint16_t qtype=0;
+	dns_read_be_u16(payload, payload_sz, payload_offset, &qtype);
+	question->qtype=(enum dns_rr_type)qtype;
+	
+	dns_read_be_u16(payload, payload_sz, payload_offset, &(question->qclass));
+
+	return DNS_DECODER_TRUE;
+}
+
+void dns_message_free(struct session *ss, void *expr_str, void *msg_free_arg)
+{
+	struct dns_message *msg=(struct dns_message *)expr_str;
+	if(msg->magic!=DNS_MESSAGE_MAGIC)
+	{
+		// abort();
+	}
+
+	if(msg->type==DNS_MESSAGE_TRANSACTION_BEGIN || msg->type==DNS_MESSAGE_TRANSACTION_END)
+	{
+		FREE(msg);
+		return ;
+	}
+
+	msg->magic=0;
+	msg->type=DNS_MESSAGE_MAX;
+	
+	msg->payload=NULL;
+	msg->payload_sz=0;
+	msg->payload_offset=0;
+
+	msg->n_question=0;
+	msg->n_answer_rr=0;
+	msg->n_authority_rr=0;
+	msg->n_additional_rr=0;
+
+	msg->n_real_answer_rr=0;
+	msg->n_real_authority_rr=0;
+	msg->n_real_additional_rr=0;
+
+	msg->answer_rr=NULL;
+	msg->authority_rr=NULL;
+	msg->additional_rr=NULL;
+
+	msg->trans_identifier_id=0;
+	msg->decode_rr_status=DNS_RR_STATUS_INIT;
+	memset(&(msg->flag), 0, sizeof(struct dns_flag));
+	memset(&(msg->question), 0, sizeof(struct dns_query_question));
+
+	msg->rr_capacity_offset=0;
+}
+
+void dns_decoder_session_transaction_add(struct dns_decoder_context *per_ss_ctx, struct dns_transaction *current_trans)
+{
+	HASH_ADD_INT(per_ss_ctx->trans_hash, message_id, current_trans);
+	DL_APPEND(per_ss_ctx->trans_list_head, current_trans);
+	per_ss_ctx->trans_list_num++;
+}
+
+void dns_decoder_session_transaction_del(struct dns_decoder_context *per_ss_ctx, struct dns_transaction *current_trans)
+{
+	DL_DELETE(per_ss_ctx->trans_list_head, current_trans);
+	per_ss_ctx->trans_list_num--;
+	
+	HASH_DEL(per_ss_ctx->trans_hash, current_trans);
+	FREE(current_trans);
+}
+
+void dns_message_transaction_publish(struct session *ss, enum dns_message_type type, int32_t topic_id, int32_t current_trans_idx)
+{
+	struct dns_message *msg=(struct dns_message *)CALLOC(struct dns_message, 1);
+	msg->magic=DNS_MESSAGE_MAGIC;
+	msg->type=type;
+	msg->current_trans_idx=current_trans_idx;
+
+	session_mq_publish_message(ss, topic_id, msg);
+}
+
+void dns_decoder_entry(struct session *ss, uint8_t *payload, size_t payload_sz, void *per_session_ctx, void *plugin_env_str)
+{
+    struct dns_header dns_hdr={0};
+    populate_dns_header(&dns_hdr, payload);
+
+    int32_t ret=validate_dns_header(&dns_hdr);
+    if(ret==DNS_DECODER_FALSE) 
+	{
+        return ;
+    }
+
+	size_t payload_offset=DNS_HEADER_SIZE;
+	int32_t thread_id=session_get_current_thread_id(ss);
+	struct dns_decoder_plugin_env *plugin_env=(struct dns_decoder_plugin_env *)plugin_env_str;
+
+	enum session_addr_type addr_type=SESSION_ADDR_TYPE_UNKNOWN;
+    session_get0_addr(ss, &addr_type);
+	const char *ip_version=(addr_type==SESSION_ADDR_TYPE_IPV4_TCP || addr_type==SESSION_ADDR_TYPE_IPV4_UDP) ? TAG_VALUE_IP_VERSION_IPV4 : TAG_VALUE_IP_VERSION_IPV6;
+	const char *ip_protocol=(addr_type==SESSION_ADDR_TYPE_IPV4_TCP || addr_type==SESSION_ADDR_TYPE_IPV6_TCP) ? TAG_VALUE_IP_PROTOCOL_TCP : TAG_VALUE_IP_PROTOCOL_UDP;
+
+	struct dns_message *data_msg=&(plugin_env->per_thread_data_msg[thread_id]);
+	data_msg->n_real_answer_rr=0;
+	data_msg->n_real_authority_rr=0;
+	data_msg->n_real_additional_rr=0;
+	data_msg->rr_capacity_offset=0;
+
+	data_msg->magic=DNS_MESSAGE_MAGIC;
+	data_msg->trans_identifier_id=dns_hdr.id;
+    dns_flag_copy(&(data_msg->flag), &dns_hdr);
+
+    data_msg->n_question=dns_hdr.qdcount;
+	data_msg->n_answer_rr=dns_hdr.ancount;
+	data_msg->n_authority_rr=dns_hdr.aucount;
+	data_msg->n_additional_rr=dns_hdr.adcount;
+
+    data_msg->type=((dns_hdr.qr==0) ? DNS_MESSAGE_QUERY : DNS_MESSAGE_RESPONSE);
+	if(data_msg->n_question==1)
+	{
+		ret=dns_query_question_decode(&(data_msg->question), payload, payload_sz, &payload_offset);
+		if(ret<DNS_DECODER_FALSE)
+		{
+			dns_message_free(ss, data_msg, NULL);
+
+			size_t tag_offset=3;
+			const char *tag_key[tag_offset]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, TAG_KEY_DECODE_FIELD};
+			const char *tag_value[tag_offset]={ip_version, ip_protocol, "question"};
+			dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_ERROR, tag_key, tag_value, tag_offset, 1, session_get_current_thread_id(ss));
+			return ;
+		}
+	}
+
+	struct dns_transaction *current_trans=NULL;
+	struct dns_decoder_context *per_ss_ctx=(struct dns_decoder_context *)per_session_ctx;
+
+	int32_t message_id=data_msg->trans_identifier_id;
+	HASH_FIND_INT(per_ss_ctx->trans_hash, &message_id, current_trans);
+	if(current_trans!=NULL && data_msg->type==DNS_MESSAGE_QUERY)
+	{
+		per_ss_ctx->trans_list_num--;
+		dns_message_transaction_publish(ss, DNS_MESSAGE_TRANSACTION_END, plugin_env->dns.topic_id, current_trans->trans_idx);
+		dns_decoder_session_transaction_del(per_ss_ctx, current_trans);
+		current_trans=NULL;
+		
+		size_t tag_offset=4;
+		const char *tag_key[tag_offset]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, TAG_KEY_MESSAGE_TYPE, TAG_KEY_MESSAGE_STATUS};
+		const char *tag_value[tag_offset]={ip_version, ip_protocol, "transaction_end", "duplicate"};
+		dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_SEND, tag_key, tag_value, tag_offset, 1, session_get_current_thread_id(ss));
+	}
+
+	if(current_trans==NULL)
+	{
+		current_trans=(struct dns_transaction *)CALLOC(struct dns_transaction, 1);
+		current_trans->trans_idx=(++per_ss_ctx->trans_count);
+		current_trans->message_id=message_id;
+		dns_decoder_session_transaction_add(per_ss_ctx, current_trans);
+		dns_message_transaction_publish(ss, DNS_MESSAGE_TRANSACTION_BEGIN, plugin_env->dns.topic_id, current_trans->trans_idx);
+
+		size_t tag_offset=4;
+		const char *tag_key[tag_offset]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, TAG_KEY_MESSAGE_TYPE, TAG_KEY_MESSAGE_STATUS};
+		const char *tag_value[tag_offset]={ip_version, ip_protocol, "transaction_begin", "normal"};
+		dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_SEND, tag_key, tag_value, tag_offset, 1, session_get_current_thread_id(ss));
+	}
+
+	data_msg->payload=payload;
+	data_msg->payload_sz=payload_sz;
+	data_msg->payload_offset=payload_offset;
+	data_msg->decode_rr_status=DNS_RR_STATUS_INIT;
+	data_msg->current_trans_idx=current_trans->trans_idx;
+
+    session_mq_publish_message(ss, plugin_env->dns.topic_id, data_msg);
+
+	size_t tag_offset=4;
+	const char *type=((data_msg->type==DNS_MESSAGE_RESPONSE) ? "response" : "query");
+	const char *tag_key[tag_offset]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, TAG_KEY_MESSAGE_TYPE, TAG_KEY_MESSAGE_STATUS};
+	const char *tag_value[tag_offset]={ip_version, ip_protocol, type, "normal"};
+	dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_SEND, tag_key, tag_value, tag_offset, 1, session_get_current_thread_id(ss));
+
+	if(data_msg->type==DNS_MESSAGE_RESPONSE)
+	{
+		dns_message_transaction_publish(ss, DNS_MESSAGE_TRANSACTION_END, plugin_env->dns.topic_id, current_trans->trans_idx);
+		dns_decoder_session_transaction_del(per_ss_ctx, current_trans);
+
+		size_t tag_offset=4;
+		const char *tag_key[tag_offset]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, TAG_KEY_MESSAGE_TYPE, TAG_KEY_MESSAGE_STATUS};
+		const char *tag_value[tag_offset]={ip_version, ip_protocol, "transaction_end", "normal"};
+		dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_SEND, tag_key, tag_value, tag_offset, 1, session_get_current_thread_id(ss));
+	}
+
+	if(per_ss_ctx->trans_list_num > plugin_env->max_cache_trans_num)
+	{
+		current_trans=per_ss_ctx->trans_list_head;
+		dns_message_transaction_publish(ss, DNS_MESSAGE_TRANSACTION_END, plugin_env->dns.topic_id, current_trans->trans_idx);
+		dns_decoder_session_transaction_del(per_ss_ctx, current_trans);
+
+		size_t tag_offset=4;
+		const char *tag_key[tag_offset]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, TAG_KEY_MESSAGE_TYPE, TAG_KEY_MESSAGE_STATUS};
+		const char *tag_value[tag_offset]={ip_version, ip_protocol, "transaction_end", "terminate"};
+		dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_SEND, tag_key, tag_value, tag_offset, 1, session_get_current_thread_id(ss));
+	}
+}
+
+void dns_udp_session_ingress_packet_cb(struct session *ss, int32_t topic_id, const void *msg, void *per_session_ctx, void *plugin_env_str)
+{
+    size_t payload_sz=0;
+    uint8_t *payload=(uint8_t *)session_get0_current_payload(ss, &payload_sz);
+    if(payload_sz<DNS_HEADER_SIZE || payload==NULL) 
+	{
+        return ;
+    }
+
+	dns_decoder_entry(ss, payload, payload_sz, per_session_ctx, plugin_env_str);
+}
+
+void dns_tcp_stream_session_segment_data_cb(struct session *ss, int32_t topic_id, const void *msg, void *per_session_ctx, void *plugin_env_str)
+{
+	size_t segment_buff_sz=0;
+	uint8_t *segment_buff=NULL;
+	segment_buff=(uint8_t *)session_get0_current_payload(ss, &segment_buff_sz);
+
+	if(segment_buff_sz==0 || segment_buff==NULL)
+	{
+		return ;
+	}
+
+	struct dns_decoder_context *per_ss_ctx=(struct dns_decoder_context *)(per_session_ctx);
+	switch(segment_buff_sz)
+	{
+		case	1:
+			if(per_ss_ctx->cache_sz>=2)
+			{
+				stellar_session_plugin_dettach_current_session(ss);
+				return ;
+			}
+			
+			per_ss_ctx->cache_ptr[per_ss_ctx->cache_sz]=segment_buff[0];
+			per_ss_ctx->cache_sz++;
+			return ;
+		case	2:
+			if(per_ss_ctx->cache_sz>0)
+			{
+				stellar_session_plugin_dettach_current_session(ss);
+				return ;
+			}
+
+			memcpy(per_ss_ctx->cache_ptr, segment_buff, segment_buff_sz);
+			per_ss_ctx->cache_sz=segment_buff_sz;
+			return ;
+		default:
+			break;
+	}
+
+	switch(per_ss_ctx->cache_sz)
+	{
+		case	0:
+			memcpy(per_ss_ctx->cache_ptr, segment_buff, 2);
+			segment_buff+=2;
+			segment_buff_sz-=2;
+			per_ss_ctx->cache_sz=2;
+			break;
+		case	1:
+			memcpy(per_ss_ctx->cache_ptr+1, segment_buff, 1);
+			segment_buff+=1;
+			segment_buff_sz-=1;
+			per_ss_ctx->cache_sz=2;
+			break;
+		case	2:
+			break;
+		default:
+			abort();
+	}
+
+	uint16_t check_sz=*(uint16_t *)(per_ss_ctx->cache_ptr);
+	if(ntohs(check_sz)!=segment_buff_sz)
+	{
+		return ;
+	}
+
+	dns_decoder_entry(ss, segment_buff, segment_buff_sz, per_session_ctx, plugin_env_str);
+	per_ss_ctx->cache_sz=0;
+}
+
+void *dns_decoder_per_session_context_new(struct session *ss, void *plugin_env_str)
+{
+	uint64_t inner_flag=0;
+    int32_t ret=session_is_innermost(ss, &inner_flag);
+    if(0==ret)
+	{
+		stellar_session_plugin_dettach_current_session(ss);
+        return NULL;
+    }
+
+	const char *ip_version=NULL;
+	const char *ip_protocol=NULL;
+	uint16_t sport=0,dport=0;
+    enum session_addr_type addr_type=SESSION_ADDR_TYPE_UNKNOWN;
+    struct session_addr *addr=session_get0_addr(ss, &addr_type);
+
+	switch(addr_type)
+	{
+		case	SESSION_ADDR_TYPE_IPV4_TCP:
+		case	SESSION_ADDR_TYPE_IPV4_UDP:
+			sport=addr->ipv4.sport;
+			dport=addr->ipv4.dport;
+			ip_version=TAG_VALUE_IP_VERSION_IPV4;
+			ip_protocol=(addr_type==SESSION_ADDR_TYPE_IPV4_TCP) ? TAG_VALUE_IP_PROTOCOL_TCP : TAG_VALUE_IP_PROTOCOL_UDP;
+			break;
+		case	SESSION_ADDR_TYPE_IPV6_TCP:
+		case	SESSION_ADDR_TYPE_IPV6_UDP:
+			sport=addr->ipv6.sport;
+			dport=addr->ipv6.dport;
+			ip_version=TAG_VALUE_IP_VERSION_IPV6;
+			ip_protocol=(addr_type==SESSION_ADDR_TYPE_IPV6_TCP) ? TAG_VALUE_IP_PROTOCOL_TCP : TAG_VALUE_IP_PROTOCOL_UDP;
+			break;
+		default:
+			stellar_session_plugin_dettach_current_session(ss);
+			return NULL;
+	}
+
+	struct dns_decoder_plugin_env *plugin_env=(struct dns_decoder_plugin_env *)plugin_env_str;
+	for(int32_t i=0; i<plugin_env->n_net_port; i++)
+	{
+		if((sport==plugin_env->net_port[i]) || (dport==plugin_env->net_port[i]))
+		{
+			const char *tag_key[3]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, "memory"};
+			const char *tag_value[3]={ip_version, ip_protocol, "ss_ctx"};
+			dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_NEW, tag_key, tag_value, 3, 1, session_get_current_thread_id(ss));
+
+			return CALLOC(struct dns_decoder_context, 1);
+		}
+	}
+
+	stellar_session_plugin_dettach_current_session(ss);
+
+	return NULL;
+}
+
+void dns_decoder_per_session_context_free(struct session *ss, void *session_ctx, void *plugin_env_str)
+{
+	if(session_ctx==NULL)
+	{
+		return ;
+	}
+
+	struct dns_decoder_context *per_ss_ctx=(struct dns_decoder_context *)session_ctx;
+	struct dns_transaction *current_trans=NULL, *tmp_trans=NULL;
+	HASH_ITER(hh, per_ss_ctx->trans_hash, current_trans, tmp_trans)
+	{
+		dns_decoder_session_transaction_del(per_ss_ctx, current_trans);
+	}
+
+	FREE(session_ctx);
+
+	struct dns_decoder_plugin_env *plugin_env=(struct dns_decoder_plugin_env *)plugin_env_str;
+	
+	enum session_addr_type addr_type=SESSION_ADDR_TYPE_UNKNOWN;
+    session_get0_addr(ss, &addr_type);
+	const char *ip_version=(addr_type==SESSION_ADDR_TYPE_IPV4_TCP || addr_type==SESSION_ADDR_TYPE_IPV4_UDP) ? TAG_VALUE_IP_VERSION_IPV4 : TAG_VALUE_IP_VERSION_IPV6;
+	const char *ip_protocol=(addr_type==SESSION_ADDR_TYPE_IPV4_TCP || addr_type==SESSION_ADDR_TYPE_IPV6_TCP) ? TAG_VALUE_IP_PROTOCOL_TCP : TAG_VALUE_IP_PROTOCOL_UDP;
+
+	const char *tag_key[3]={TAG_KEY_IP_VERSION, TAG_KEY_IP_PROTOCOL, "memory"};
+	const char *tag_value[3]={ip_version, ip_protocol, "ss_ctx"};
+	dns_decoder_local_file_counter_incby(plugin_env, LOCAL_STAT_COUNTER_FREE, tag_key, tag_value, 3, 1, session_get_current_thread_id(ss));
+}
+
+int32_t dns_decoder_config_load(const char *cfg_path, struct dns_decoder_plugin_env *plugin_env)
+{
+    FILE *fp=fopen(cfg_path, "r");
+    if (NULL==fp)
+    {
+        fprintf(stderr, "[%s:%d] Can't open config file: %s", __FUNCTION__, __LINE__, cfg_path);
+        return -1;
+    }
+
+    int32_t ret=0;
+    char errbuf[256]={0};
+
+    toml_table_t *root=toml_parse_file(fp, errbuf, sizeof(errbuf));
+    fclose(fp);
+
+    toml_table_t *decoder_tbl=toml_table_in(root, "decoder");
+    if(NULL==decoder_tbl)
+    {
+        fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns]", __FUNCTION__, __LINE__, cfg_path);
+        toml_free(root);
+        return -1;
+    }
+
+    toml_table_t *dns_tbl=toml_table_in(decoder_tbl, "dns");
+    if(NULL==dns_tbl)
+    {
+        fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns]", __FUNCTION__, __LINE__, cfg_path);
+        toml_free(root);
+        return -1;
+    }
+
+	toml_array_t *port_array=toml_array_in(dns_tbl, "port");
+	if(NULL==port_array)
+	{
+		fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.port]", __FUNCTION__, __LINE__, cfg_path);
+		toml_free(root);
+		return -1;
+	}
+
+	//toml_array_type
+	char port_array_type=toml_array_type(port_array);
+	if(port_array_type!='i')
+	{
+		fprintf(stderr, "[%s:%d] config file: %s key: [decoder.dns.port] type is not integer", __FUNCTION__, __LINE__, cfg_path);
+		toml_free(root);
+		return -1;
+	}
+
+	plugin_env->n_net_port=toml_array_nelem(port_array);
+	plugin_env->net_port=(uint16_t *)CALLOC(uint16_t, plugin_env->n_net_port);
+	for(int32_t i=0; i<plugin_env->n_net_port; i++)
+	{
+		toml_datum_t int_val=toml_int_at(port_array, i);
+		if(int_val.ok==0)
+		{
+			fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.port[%d]]", __FUNCTION__, __LINE__, cfg_path, i);
+			ret=-1;
+			break;
+		}
+
+		plugin_env->net_port[i]=ntohs(int_val.u.i);
+	}
+
+    toml_table_t *limited_tbl=toml_table_in(dns_tbl, "limited");
+    if(NULL==limited_tbl)
+    {
+        fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.limited]", __FUNCTION__, __LINE__, cfg_path);
+        toml_free(root);
+        return -1;
+    }
+
+	toml_datum_t max_rr_num_val=toml_int_in(limited_tbl, "max_rr_num");
+	if(max_rr_num_val.ok==0)
+	{
+		fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.limited.max_rr_num]", __FUNCTION__, __LINE__, cfg_path);
+		ret=-1;
+	}
+	else
+	{
+		plugin_env->max_rr_num=max_rr_num_val.u.i;
+	}
+
+	// max_cache_trans_num
+	toml_datum_t max_cache_trans_num_val=toml_int_in(limited_tbl, "max_cache_trans_num");
+	if(max_cache_trans_num_val.ok==0)
+	{
+		fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.limited.max_cache_trans_num]", __FUNCTION__, __LINE__, cfg_path);
+		ret=-1;
+	}
+	else
+	{
+		plugin_env->max_cache_trans_num=max_cache_trans_num_val.u.i;
+	}
+
+    toml_table_t *local_stat_tbl=toml_table_in(dns_tbl, "local_stat");
+    if(NULL==local_stat_tbl)
+    {
+        fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.local_stat]", __FUNCTION__, __LINE__, cfg_path);
+        toml_free(root);
+        return -1;
+    }
+
+	toml_datum_t stat_interval_time_s_val=toml_int_in(local_stat_tbl, "stat_interval_time_s");
+	if(stat_interval_time_s_val.ok==0)
+	{
+		plugin_env->stat.interval_second=5;
+		fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.local_stat.stat_interval_time_s]", __FUNCTION__, __LINE__, cfg_path);
+	}
+	else
+	{
+		plugin_env->stat.interval_second=stat_interval_time_s_val.u.i;
+	}
+
+	toml_datum_t stat_per_thread_enable_val=toml_string_in(local_stat_tbl, "stat_per_thread_enable");
+	if(stat_per_thread_enable_val.ok==0)
+	{
+		plugin_env->stat.per_thread_enable=DNS_DECODER_FALSE;
+		fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.local_stat.stat_per_thread_enable]", __FUNCTION__, __LINE__, cfg_path);
+	}
+	else
+	{
+		if(memcmp("no", stat_per_thread_enable_val.u.s, strlen("no"))==0)
+		{
+			plugin_env->stat.per_thread_enable=DNS_DECODER_FALSE;
+		}
+		else if(memcmp("yes", stat_per_thread_enable_val.u.s, strlen("yes"))==0)
+		{
+			plugin_env->stat.per_thread_enable=DNS_DECODER_TRUE;
+		}
+		else
+		{
+			plugin_env->stat.per_thread_enable=DNS_DECODER_FALSE;
+			fprintf(stderr, "[%s:%d] config file: %s key: [decoder.dns.local_stat.stat_per_thread_enable] value is not yes or no", __FUNCTION__, __LINE__, cfg_path);
+		}
+	}
+
+	toml_datum_t name=toml_string_in(local_stat_tbl, "stat_name");
+	if(name.ok==0)
+	{
+		memcpy(plugin_env->stat.name, "DNS_DECODER", MIN(sizeof(plugin_env->stat.name)-1, strlen("DNS_DECODER")));
+		fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.local_stat.stat_name]", __FUNCTION__, __LINE__, cfg_path);
+	}
+	else
+	{
+		strncpy(plugin_env->stat.name, name.u.s, sizeof(plugin_env->stat.name));
+	}
+
+	toml_datum_t output_path=toml_string_in(local_stat_tbl, "stat_output");
+	if(output_path.ok==0)
+	{
+		memcpy(plugin_env->stat.path, "metrics/dns_decoder_local_stat.json", MIN(sizeof(plugin_env->stat.path)-1, strlen("metrics/dns_decoder_local_stat.json")));
+		fprintf(stderr, "[%s:%d] config file: %s has no key: [decoder.dns.local_stat.stat_output]", __FUNCTION__, __LINE__, cfg_path);
+	}
+	else
+	{
+		strncpy(plugin_env->stat.path, output_path.u.s, sizeof(plugin_env->stat.path));
+	}
+
+	toml_free(root);
+	
+    return ret;
+}
+
+void dns_decoder_local_file_stat_init(struct dns_decoder_plugin_env *plugin_env)
+{
+    if(plugin_env->stat.interval_second==0)
+    {
+        printf("dns_decoder_local_file_stat_init, Disable local stat, name: %s output: %s", plugin_env->stat.name, plugin_env->stat.path);
+        return ;
+    }
+
+    plugin_env->stat.fse=fieldstat_easy_new(stellar_get_worker_thread_num(plugin_env->st), plugin_env->stat.name, NULL, 0);
+    if(plugin_env->stat.fse==NULL)
+    {
+        printf("dns_decoder_local_file_stat_init, fieldstat_easy_new failed, name: %s output: %s", plugin_env->stat.name, plugin_env->stat.path);
+        exit(-1);
+    }
+
+    fieldstat_easy_enable_auto_output(plugin_env->stat.fse, plugin_env->stat.path, plugin_env->stat.interval_second);
+
+    const char *local_stat_name[LOCAL_STAT_COUNTER_MAX]={0};
+    local_stat_name[LOCAL_STAT_COUNTER_UNKNOWN]="unknown";
+	local_stat_name[LOCAL_STAT_COUNTER_SESSION]="session";
+    local_stat_name[LOCAL_STAT_COUNTER_PACKETS]="packets";
+    local_stat_name[LOCAL_STAT_COUNTER_BYTES]="bytes";
+	local_stat_name[LOCAL_STAT_COUNTER_SEND]="send";
+	local_stat_name[LOCAL_STAT_COUNTER_RECV]="recv";
+	local_stat_name[LOCAL_STAT_COUNTER_NEW]="new";
+	local_stat_name[LOCAL_STAT_COUNTER_FREE]="free";
+	local_stat_name[LOCAL_STAT_COUNTER_OK]="ok";
+	local_stat_name[LOCAL_STAT_COUNTER_ERROR]="error";
+
+
+    plugin_env->stat.metric_id=(int *)CALLOC(int, LOCAL_STAT_COUNTER_MAX);
+    for(int i=0; i<LOCAL_STAT_COUNTER_MAX; i++)
+    {
+        plugin_env->stat.metric_id[i]=fieldstat_easy_register_counter(plugin_env->stat.fse, local_stat_name[i]);
+        if(plugin_env->stat.metric_id[i]<0)
+        {
+            printf("dns_decoder_local_file_stat_init, fieldstat_easy_register_counter failed, name: %s", local_stat_name[i]);
+            exit(-1);
+        }
+    }
+}
+
+extern "C" void *dns_decoder_init(struct stellar *st)
+{
+    struct dns_decoder_plugin_env *plugin_env=CALLOC(struct dns_decoder_plugin_env, 1);
+    plugin_env->st=st;
+
+    plugin_env->plugin_id=stellar_session_plugin_register(st, dns_decoder_per_session_context_new, dns_decoder_per_session_context_free, plugin_env);
+    if(plugin_env->plugin_id<0) 
+	{
+		printf("dns_decoder_init: stellar_session_plugin_register failed\n");
+        exit(0);
+    }
+
+	dns_decoder_config_load(DNS_DECODER_TOML_PATH, plugin_env);
+
+	dns_decoder_local_file_stat_init(plugin_env);
+
+	int32_t thread_count=stellar_get_worker_thread_num(plugin_env->st);
+	plugin_env->per_thread_data_msg=(struct dns_message *)CALLOC(struct dns_message, thread_count);
+	for(int32_t i=0; i<thread_count; i++)
+	{
+		plugin_env->per_thread_data_msg[i].rr_capacity_sz=plugin_env->max_rr_num;
+		plugin_env->per_thread_data_msg[i].rr_capacity=(struct dns_resource_record *)CALLOC(struct dns_resource_record, plugin_env->max_rr_num);
+	}
+
+	plugin_env->dns.free_cb=dns_message_free;
+	plugin_env->dns.on_cb=NULL;
+	plugin_env->dns.topic_name=DNS_MESSAGE_TOPIC;
+    plugin_env->dns.topic_id=stellar_session_mq_get_topic_id(st, plugin_env->dns.topic_name);
+    if(plugin_env->dns.topic_id<0)
+	{
+        plugin_env->dns.topic_id=stellar_session_mq_create_topic(st, plugin_env->dns.topic_name, dns_message_free, NULL);
+    }
+
+	plugin_env->udp.free_cb=NULL;
+	plugin_env->udp.on_cb=dns_udp_session_ingress_packet_cb;
+	plugin_env->udp.topic_name=TOPIC_UDP;
+	plugin_env->udp.topic_id=stellar_session_mq_get_topic_id(plugin_env->st, plugin_env->udp.topic_name);
+	plugin_env->udp.sub_id=stellar_session_mq_subscribe(plugin_env->st, plugin_env->udp.topic_id, plugin_env->udp.on_cb, plugin_env->plugin_id);
+
+	plugin_env->tcp_stream.free_cb=NULL;
+	plugin_env->tcp_stream.on_cb=dns_tcp_stream_session_segment_data_cb;
+	plugin_env->tcp_stream.topic_name=TOPIC_TCP_STREAM;
+	plugin_env->tcp_stream.topic_id=stellar_session_mq_get_topic_id(plugin_env->st, plugin_env->tcp_stream.topic_name);
+	plugin_env->tcp_stream.sub_id=stellar_session_mq_subscribe(plugin_env->st, plugin_env->tcp_stream.topic_id, plugin_env->tcp_stream.on_cb, plugin_env->plugin_id);
+
+    printf("dns_decoder_init: plugin_id: %d, topic: [{name: %s -> id: %d}, {name: %s -> id: %d}, {name: %s -> id: %d}] \n",
+			plugin_env->plugin_id, 
+			plugin_env->dns.topic_name, plugin_env->dns.topic_id,
+			plugin_env->udp.topic_name, plugin_env->udp.topic_id,
+			plugin_env->tcp_stream.topic_name, plugin_env->tcp_stream.topic_id
+		);
+
+    return plugin_env;
+}
+
+extern "C" void dns_decoder_exit(void *plugin_env_str)
+{
+    if(NULL==plugin_env_str) 
+	{
+        return;
+    }
+
+    struct dns_decoder_plugin_env *plugin_env=(struct dns_decoder_plugin_env *)plugin_env_str;
+    if(plugin_env->dns.topic_id>=0) 
+	{
+        stellar_session_mq_destroy_topic(plugin_env->st, plugin_env->dns.topic_id);
+    	plugin_env->dns.topic_id=-1;
+    }
+
+	if(plugin_env->udp.topic_id>=0)
+	{
+		stellar_session_mq_destroy_topic(plugin_env->st, plugin_env->udp.topic_id);
+		plugin_env->udp.topic_id=-1;
+	}
+
+	if(plugin_env->tcp_stream.topic_id>=0)
+	{
+		stellar_session_mq_destroy_topic(plugin_env->st, plugin_env->tcp_stream.topic_id);
+		plugin_env->tcp_stream.topic_id=-1;
+	}
+
+	if(plugin_env->per_thread_data_msg!=NULL)
+	{
+		int32_t thread_count=stellar_get_worker_thread_num(plugin_env->st);
+		for(int32_t i=0; i<thread_count; i++)
+		{
+			FREE(plugin_env->per_thread_data_msg[i].rr_capacity);
+		}
+		
+		plugin_env->max_rr_num=0;
+		plugin_env->per_thread_data_msg=NULL;
+
+		FREE(plugin_env->per_thread_data_msg);
+		plugin_env->per_thread_data_msg=NULL;
+	}
+
+	// if(plugin_env->per_thread_trans_new!=NULL)
+	// {
+	// 	FREE(plugin_env->per_thread_trans_new);
+	// 	plugin_env->per_thread_trans_new=NULL;
+	// }
+
+    FREE(plugin_env_str);
+}
+
+enum dns_message_type dns_message_type_get(struct dns_message *msg)
+{
+    if (NULL==msg) {
+        return DNS_MESSAGE_MAX;
+    }
+
+    return msg->type;
+}
+
+int32_t dns_message_header_id_get(struct dns_message *msg)
+{
+	if (NULL==msg || msg->magic!=DNS_MESSAGE_MAGIC)
+	{
+		return -1;
+	}
+
+	return msg->trans_identifier_id;
+}
+
+struct dns_flag *dns_message_header_flag_get0(struct dns_message *msg)
+{
+	if (NULL==msg || msg->magic!=DNS_MESSAGE_MAGIC)
+	{
+		return NULL;
+	}
+
+	return &(msg->flag);
+}
+
+void dns_message_question_get0(struct dns_message *msg, struct dns_query_question **question, uint16_t *n_question)
+{
+	if(msg==NULL || msg->magic!=DNS_MESSAGE_MAGIC)
+	{
+		return ;
+	}
+
+	(*question)=&(msg->question);
+	(*n_question)=msg->n_question;
+}
+
+int32_t dns_resource_record_buff_get0(struct dns_resource_record *rr_capacity, int32_t rr_capacity_sz, int32_t *rr_capacity_offset, struct dns_resource_record **rr, uint16_t n_rr, uint16_t *n_real_rr)
+{
+	if(n_rr==0)
+	{
+		(*rr)=NULL;
+		(*n_real_rr)=0;
+		return DNS_DECODER_TRUE;
+	}
+
+	if(rr_capacity_sz<=(*rr_capacity_offset))
+	{
+		rr=NULL;
+		(*n_real_rr)=0;
+		return DNS_DECODER_FALSE;
+	}
+
+	(*rr)=rr_capacity+(*rr_capacity_offset);
+	(*n_real_rr)=MIN(rr_capacity_sz-(*rr_capacity_offset), n_rr);
+
+	(*rr_capacity_offset)+=(*n_real_rr);
+
+	return DNS_DECODER_TRUE;
+}
+
+int32_t dns_message_resource_record_serialize(struct dns_message *msg)
+{
+	switch(msg->decode_rr_status)
+	{
+		case	DNS_RR_STATUS_INIT:
+			break;
+		case	DNS_RR_STATUS_SUCCESS:
+			return DNS_DECODER_TRUE;
+		case	DNS_RR_STATUS_FAILURE:
+			return DNS_DECODER_FALSE;
+		default:
+			abort();
+	}
+
+	if(msg->n_answer_rr>0)
+	{
+		dns_resource_record_buff_get0(msg->rr_capacity, msg->rr_capacity_sz, &(msg->rr_capacity_offset), &(msg->answer_rr), msg->n_answer_rr, &(msg->n_real_answer_rr));
+		int32_t ret=dns_resource_record_decode(msg->payload, msg->payload_sz, &(msg->payload_offset), msg->answer_rr, msg->n_real_answer_rr);
+		if(ret==DNS_DECODER_FALSE)
+		{
+			msg->answer_rr=NULL;
+			msg->n_real_answer_rr=0;
+			msg->decode_rr_status=DNS_RR_STATUS_FAILURE;
+			return DNS_DECODER_FALSE;
+		}
+	}
+
+	if(msg->n_authority_rr>0)
+	{
+		dns_resource_record_buff_get0(msg->rr_capacity, msg->rr_capacity_sz, &(msg->rr_capacity_offset), &(msg->authority_rr), msg->n_authority_rr, &(msg->n_real_authority_rr));
+		int32_t ret=dns_resource_record_decode(msg->payload, msg->payload_sz, &(msg->payload_offset), msg->authority_rr, msg->n_real_authority_rr);
+		if(ret==DNS_DECODER_FALSE)
+		{
+			msg->authority_rr=NULL;
+			msg->n_real_authority_rr=0;
+			msg->decode_rr_status=DNS_RR_STATUS_FAILURE;
+			return DNS_DECODER_FALSE;
+		}
+	}
+
+	if(msg->n_additional_rr>0)
+	{
+		dns_resource_record_buff_get0(msg->rr_capacity, msg->rr_capacity_sz, &(msg->rr_capacity_offset), &(msg->additional_rr), msg->n_additional_rr, &(msg->n_real_additional_rr));
+		int32_t ret=dns_resource_record_decode(msg->payload, msg->payload_sz, &(msg->payload_offset), msg->additional_rr, msg->n_real_additional_rr);
+		if(ret==DNS_DECODER_FALSE)
+		{
+			msg->additional_rr=NULL;
+			msg->n_real_additional_rr=0;
+			msg->decode_rr_status=DNS_RR_STATUS_FAILURE;
+			return DNS_DECODER_FALSE;
+		}
+	}
+
+	msg->decode_rr_status=DNS_RR_STATUS_SUCCESS;
+	return DNS_DECODER_TRUE;
+}
+
+void dns_message_answer_resource_record_get0(struct dns_message *msg, struct dns_resource_record **answer_rr, uint16_t *n_answer_rr)
+{
+	if(msg==NULL || msg->magic!=DNS_MESSAGE_MAGIC)
+	{
+		return ;
+	}
+
+	int32_t ret=dns_message_resource_record_serialize(msg);
+	(*answer_rr)=((ret==DNS_DECODER_TRUE) ? msg->answer_rr : NULL);
+	(*n_answer_rr)=((ret==DNS_DECODER_TRUE) ? msg->n_real_answer_rr : 0);
+}
+
+void dns_message_authority_resource_record_get0(struct dns_message *msg, struct dns_resource_record **authority_rr, uint16_t *n_authority_rr)
+{
+	if(msg==NULL || msg->magic!=DNS_MESSAGE_MAGIC)
+	{
+		return ;
+	}
+
+	int32_t ret=dns_message_resource_record_serialize(msg);
+	(*authority_rr)=((ret==DNS_DECODER_TRUE) ? msg->authority_rr : NULL);
+	(*n_authority_rr)=((ret==DNS_DECODER_TRUE) ? msg->n_real_authority_rr : 0);
+}
+
+void dns_message_additional_resource_record_get0(struct dns_message *msg, struct dns_resource_record **additional_rr, uint16_t *n_additional_rr)
+{
+	if(msg==NULL || msg->magic!=DNS_MESSAGE_MAGIC)
+	{
+		return ;
+	}
+
+	int32_t ret=dns_message_resource_record_serialize(msg);
+	(*additional_rr)=((ret==DNS_DECODER_TRUE) ? msg->additional_rr : NULL);
+	(*n_additional_rr)=((ret==DNS_DECODER_TRUE) ? msg->n_real_additional_rr : 0);
+}