↩ revert(packet mq & exdata): remove correlative code

8 files changed, 1 insertions, 938 deletions

diff --git a/examples/stellar_plugin/simple_stellar_plugin.c b/examples/stellar_plugin/simple_stellar_plugin.c
index 95f078e..467a38b 100644
--- a/examples/stellar_plugin/simple_stellar_plugin.c
+++ b/examples/stellar_plugin/simple_stellar_plugin.c
@@ -3,8 +3,6 @@
 #include "stellar/utils.h"
 #include "stellar/session_exdata.h"
 #include "stellar/session_mq.h"
-#include "stellar/packet_exdata.h"
-#include "stellar/packet_mq.h"
 
 #include <stdio.h>
 #include <string.h>
@@ -16,8 +14,6 @@ struct simple_stellar_plugin_env
 	struct stellar *st;
     int session_plugin_id;
     int session_exdata_idx;
-    int packet_exdata_idx;
-	int packet_topic_id;
 	int stat_topic_id;
 	int egress_topic_id;
 	int tcp_topic_id;
@@ -134,8 +130,6 @@ static void simple_plugin_on_session_msg(struct session *sess, int topic_id, con
 void simple_plugin_on_packet(struct packet *pkt,  unsigned char ip_protocol, void *plugin_env)
 {
 	struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
-	packet_exdata_set(pkt, env->packet_exdata_idx, env);
-	packet_mq_publish_message(pkt, env->packet_topic_id, env);
 	switch (ip_protocol)
 	{
 	case IPPROTO_TCP:
@@ -158,14 +152,6 @@ void simple_plugin_on_packet(struct packet *pkt,  unsigned char ip_protocol, voi
 	return;
 }
 
-void simple_plugin_packet_get_exdata(struct packet *pkt,  unsigned char ip_protocol, void *plugin_env)
-{
-	struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
-	struct simple_stellar_plugin_env *exdata = (struct simple_stellar_plugin_env *)packet_exdata_get(pkt, env->packet_exdata_idx);
-	assert(memcmp(env, exdata, sizeof(struct simple_stellar_plugin_env)) == 0);
-	return;
-}
-
 int simple_plugin_on_polling(void *plugin_env)
 {
 	struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
@@ -217,30 +203,6 @@ void *simple_session_packet_plugin_init(struct stellar *st)
 		exit(-1);
 	}
 
-	env->packet_topic_id=stellar_packet_mq_get_topic_id(st, "TOPIC_PACKET_ENV");
-	if(env->packet_topic_id < 0)
-	{
-		env->packet_topic_id=stellar_packet_mq_create_topic(st, "TOPIC_PACKET_ENV", simple_plugin_packet_msg_free, env);
-	}
-
-	tcp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_TCP, simple_plugin_packet_get_exdata, env);	
-	udp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_UDP, simple_plugin_packet_get_exdata, env);	
-	icmp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_ICMP, simple_plugin_packet_get_exdata, env);	
-	icmp6_plugin_id=stellar_packet_plugin_register(st, IPPROTO_ICMPV6, simple_plugin_packet_get_exdata, env);	
-
-	if(tcp_plugin_id < 0 || udp_plugin_id < 0 || icmp_plugin_id < 0 || icmp6_plugin_id < 0)
-	{
-		perror("register packet plugin get exdata return invalid plugin id\n");
-		exit(-1);
-	}
-
-	stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, tcp_plugin_id);
-	stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, udp_plugin_id);
-	stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, icmp_plugin_id);
-	stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, icmp6_plugin_id);
-
-	env->packet_exdata_idx=stellar_packet_exdata_new_index(st, "EXDATA_PACKET_ENV", simple_plugin_packet_exdata_free, env);
-
 	int polling_plugin_id=stellar_polling_plugin_register(st, simple_plugin_on_polling, env);
 	if(polling_plugin_id < 0)
 	{
diff --git a/include/stellar/packet_exdata.h b/include/stellar/packet_exdata.h
deleted file mode 100644
index 7620eae..0000000
--- a/include/stellar/packet_exdata.h
+++ /dev/null
@@ -1,8 +0,0 @@
-#pragma once
-
-#include "stellar.h"
-
-typedef void packet_exdata_free(struct packet *pkt, int idx, void *ex_ptr, void *arg);
-int stellar_packet_exdata_new_index(struct stellar *st, const char *name, packet_exdata_free *free_func,void *arg); 
-int packet_exdata_set(struct packet *pkt, int idx, void *ex_ptr);
-void *packet_exdata_get(struct packet *pkt, int idx);
\ No newline at end of file
diff --git a/include/stellar/packet_mq.h b/include/stellar/packet_mq.h
deleted file mode 100644
index 94bb39b..0000000
--- a/include/stellar/packet_mq.h
+++ /dev/null
@@ -1,22 +0,0 @@
-#pragma once
-
-#include "stellar.h"
-
-//session mq
-typedef void packet_msg_free_cb_func(struct packet *pkt, void *msg, void *msg_free_arg);
-typedef void on_packet_msg_cb_func(struct packet *pkt, int topic_id, const void *msg, void *plugin_env);
-
-//return topic_id
-int stellar_packet_mq_create_topic(struct stellar *st, const char *topic_name, packet_msg_free_cb_func *msg_free_cb, void *msg_free_arg);
-
-int stellar_packet_mq_get_topic_id(struct stellar *st, const char *topic_name);
-
-int stellar_packet_mq_update_topic(struct stellar *st, int topic_id, packet_msg_free_cb_func *msg_free_cb, void *msg_free_arg);
-
-int stellar_packet_mq_destroy_topic(struct stellar *st, int topic_id);
-
-//return 0 if success, otherwise return -1.
-int stellar_packet_mq_subscribe(struct stellar *st, int topic_id, on_packet_msg_cb_func *plugin_on_msg_cb, int plugin_id); //packet plugin only
-
-int packet_mq_publish_message(struct packet *pkt, int topic_id, void *msg);
-
diff --git a/src/plugin_manager/plugin_manager.c b/src/plugin_manager/plugin_manager.c
index 9f9f7bd..3eeeb21 100644
--- a/src/plugin_manager/plugin_manager.c
+++ b/src/plugin_manager/plugin_manager.c
@@ -70,32 +70,6 @@ PLUGIN_SPEC_LOAD_ERROR:
 	return NULL;	
 }
 
-
-static struct plugin_manger_per_thread_data *plugin_manager_per_thread_data_new(struct stellar *st)
-{
-	if(st == NULL)return NULL;
-	int thread_num=stellar_get_worker_thread_num(st);
-	struct plugin_manger_per_thread_data *per_thread_data = CALLOC(struct plugin_manger_per_thread_data, thread_num);
-	return per_thread_data;
-}
-
-
-static void plugin_manager_per_thread_data_free(struct plugin_manger_per_thread_data *per_thread_data, struct stellar *st)
-{
-	if(per_thread_data == NULL || st == NULL)return;
-	int thread_num=stellar_get_worker_thread_num(st);
-	struct plugin_manger_per_thread_data *p_data;
-	for (int i = 0; i < thread_num; i++)
-	{
-		p_data=per_thread_data+i;
-		if(p_data->per_thread_pkt_exdata_array.exdata_array)FREE(p_data->per_thread_pkt_exdata_array.exdata_array);
-		if(p_data->per_thread_pkt_mq.mq)FREE(p_data->per_thread_pkt_mq.mq);
-	}
-	FREE(per_thread_data);
-	return;
-}
-
-
 struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char *plugin_spec_file_path)
 {
 	int spec_num;
@@ -131,7 +105,6 @@ struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char
 		}
 	}
 	FREE(specs);
-	plug_mgr->per_thread_data = plugin_manager_per_thread_data_new(st);
 	return plug_mgr;
 }
 
@@ -156,16 +129,6 @@ void plugin_manager_exit(struct plugin_manager_schema *plug_mgr)
 		}
 		utarray_free(plug_mgr->session_mq_schema_array);
 	}
-    if(plug_mgr->packet_mq_schema_array)
-	{
-		for(unsigned int i = 0; i < utarray_len(plug_mgr->packet_mq_schema_array); i++)
-		{
-			stellar_packet_mq_destroy_topic(plug_mgr->st, i);
-		}
-		utarray_free(plug_mgr->packet_mq_schema_array);
-	}	
-
-	if(plug_mgr->packet_exdata_schema_array)utarray_free(plug_mgr->packet_exdata_schema_array);
 	if(plug_mgr->session_exdata_schema_array)utarray_free(plug_mgr->session_exdata_schema_array);
 	if(plug_mgr->registered_polling_plugin_array)utarray_free(plug_mgr->registered_polling_plugin_array);
 	if(plug_mgr->registered_packet_plugin_array)
@@ -186,7 +149,6 @@ void plugin_manager_exit(struct plugin_manager_schema *plug_mgr)
         }
         utarray_free(plug_mgr->registered_session_plugin_array);
 	}
-	plugin_manager_per_thread_data_free(plug_mgr->per_thread_data, plug_mgr->st);
 	FREE(plug_mgr);
 	return;
 }
@@ -261,66 +223,6 @@ void *stellar_exdata_get(UT_array *exdata_schema, struct stellar_exdata *exdata_
 }
 
 /*******************************
- * PACKET EXDATA *
- *******************************/
-
-
-static struct stellar_exdata *per_thread_packet_exdata_arrary_get(struct plugin_manager_schema *plug_mgr)
-{
-	if(plug_mgr==NULL || plug_mgr->packet_exdata_schema_array == NULL)return NULL;
-	int tid=stellar_get_current_thread_id(plug_mgr->st);
-	if(plug_mgr->per_thread_data[tid].per_thread_pkt_exdata_array.exdata_array == NULL)
-	{
-		unsigned int len = utarray_len(plug_mgr->packet_exdata_schema_array);
-		plug_mgr->per_thread_data[tid].per_thread_pkt_exdata_array.exdata_array = CALLOC(struct stellar_exdata, len);
-	}
-	return plug_mgr->per_thread_data[tid].per_thread_pkt_exdata_array.exdata_array;	
-}
-
-static void per_thread_packet_exdata_arrary_clean(struct plugin_manager_schema *plug_mgr, struct packet *pkt)
-{
-	if(plug_mgr==NULL || plug_mgr->packet_exdata_schema_array == NULL)return;
-	unsigned int len=utarray_len(plug_mgr->packet_exdata_schema_array);
-	struct stellar_exdata *per_thread_pkt_exdata_arrary = per_thread_packet_exdata_arrary_get(plug_mgr);
-	if(per_thread_pkt_exdata_arrary == NULL)return;
-	for (unsigned int i = 0; i < len; i++)
-	{
-		void *exdata = (per_thread_pkt_exdata_arrary + i)->exdata;
-		struct stellar_exdata_schema *schema = (struct stellar_exdata_schema *)utarray_eltptr(plug_mgr->packet_exdata_schema_array, i);
-		if (exdata)
-		{
-			if (schema->pkt_free_func)
-			{
-				schema->pkt_free_func(pkt, i, exdata, schema->free_arg);
-			}
-			(per_thread_pkt_exdata_arrary + i)->exdata=NULL;
-		}
-	}
-}
-
-int stellar_packet_exdata_new_index(struct stellar *st, const char *name, packet_exdata_free *free_func,void *free_arg) 
-{
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	assert(plug_mgr);
-	return stellar_exdata_new_index(st, name, &plug_mgr->packet_exdata_schema_array, (void*)free_func, free_arg);
-}
-
-int packet_exdata_set(struct packet *pkt, int idx, void *ex_ptr)
-{
-	if(pkt == NULL)return -1;
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(packet_stellar_get(pkt));
-	return stellar_exdata_set(plug_mgr->packet_exdata_schema_array, per_thread_packet_exdata_arrary_get(plug_mgr), idx, ex_ptr);
-}
-
-void *packet_exdata_get(struct packet *pkt, int idx)
-{
-	if(pkt == NULL)return NULL;
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(packet_stellar_get(pkt));
-	return stellar_exdata_get( plug_mgr->packet_exdata_schema_array, per_thread_packet_exdata_arrary_get(plug_mgr), idx);
-}
-
-
-/*******************************
  * SESSION EXDATA *
  *******************************/
 
@@ -462,184 +364,6 @@ int stellar_mq_publish_message(int topic_id, void *data, UT_array *mq_schema_arr
 
 UT_icd stellar_mq_subscriber_info_icd = {sizeof(struct stellar_mq_subscriber_info), NULL, NULL, NULL};
 
-/*******************************
- * PACKET MQ *
- *******************************/
-
-static inline int stellar_current_thread_packet_mq_counter_inc(struct plugin_manager_schema *plug_mgr)
-{
-	if(plug_mgr==NULL)return -1;
-	int tid = stellar_get_current_thread_id(plug_mgr->st);
-	plug_mgr->per_thread_data[tid].per_thread_pkt_mq.pub_msg_cnt+=1;
-	return plug_mgr->per_thread_data[tid].per_thread_pkt_mq.pub_msg_cnt;
-}
-
-static inline void stellar_current_thread_packet_mq_counter_reset(struct plugin_manager_schema *plug_mgr)
-{
-	if(plug_mgr==NULL)return;
-	int tid = stellar_get_current_thread_id(plug_mgr->st);
-	plug_mgr->per_thread_data[tid].per_thread_pkt_mq.pub_msg_cnt=0;
-}
-
-static inline int stellar_current_thread_packet_mq_counter_get(struct plugin_manager_schema *plug_mgr)
-{
-	if(plug_mgr==NULL)return 0;
-	int tid = stellar_get_current_thread_id(plug_mgr->st);
-	return plug_mgr->per_thread_data[tid].per_thread_pkt_mq.pub_msg_cnt;
-}
-
-int stellar_packet_mq_create_topic(struct stellar *st, const char *topic_name, packet_msg_free_cb_func *msg_free_cb, void *msg_free_arg)
-{
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	assert(plug_mgr);
-	int topic_id=stellar_mq_create_topic(st, topic_name, (void *)msg_free_cb, msg_free_arg, &plug_mgr->packet_mq_schema_array);
-	if(topic_id>=0)plug_mgr->packet_mq_topic_num+=1;
-	return topic_id;
-}
-
-int stellar_packet_mq_get_topic_id(struct stellar *st, const char *topic_name)
-{
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	assert(plug_mgr);
-	return stellar_mq_get_topic_id(topic_name, plug_mgr->packet_mq_schema_array);	
-}
-
-int stellar_packet_mq_update_topic(struct stellar *st, int topic_id, packet_msg_free_cb_func *msg_free_cb, void *msg_free_arg)
-{
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	assert(plug_mgr);
-	return stellar_mq_update_topic(topic_id, (void *)msg_free_cb, msg_free_arg, plug_mgr->packet_mq_schema_array);	
-}
-
-int stellar_packet_mq_destroy_topic(struct stellar *st, int topic_id)
-{
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	assert(plug_mgr);
-	int ret = stellar_mq_destroy_topic(topic_id, plug_mgr->packet_mq_schema_array);	
-	if(ret==1)plug_mgr->packet_mq_topic_num-=1;
-	return ret;
-}
-
-//return 0 if success, otherwise return -1.
-int stellar_packet_mq_subscribe(struct stellar *st, int topic_id, on_packet_msg_cb_func *plugin_on_msg_cb, int plugin_id) //packet plugin only
-{
-	if(plugin_id < PACKET_PULGIN_ID_BASE || plugin_id >= POLLING_PULGIN_ID_BASE)return -1;// ignore session or polling  plugin
-	int plugin_idx=plugin_id-PACKET_PULGIN_ID_BASE;
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	if(plug_mgr == NULL || plug_mgr->packet_mq_schema_array==NULL || plug_mgr->registered_packet_plugin_array == NULL)return -1;
-	
-    unsigned int len = utarray_len(plug_mgr->packet_mq_schema_array);
-    if (len <= (unsigned int)topic_id)return -1;
-	struct registered_packet_plugin_schema *packet_plugin_schema = (struct registered_packet_plugin_schema *)utarray_eltptr(plug_mgr->registered_packet_plugin_array, (unsigned)plugin_idx);
-	if(packet_plugin_schema==NULL)return -1;
-
-    struct stellar_mq_topic_schema *topic = (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->packet_mq_schema_array, (unsigned int)topic_id);
-	if(topic==NULL)return -1;
-
-	if(packet_plugin_schema->registed_packet_mq_subscriber_info==NULL)
-	{
-		utarray_new(packet_plugin_schema->registed_packet_mq_subscriber_info, &stellar_mq_subscriber_info_icd);
-	}
-
-	// if plugin already subscribe current topic, return 0
-	struct stellar_mq_subscriber_info *p=NULL;
-  	while( (p=(struct stellar_mq_subscriber_info *)utarray_next(packet_plugin_schema->registed_packet_mq_subscriber_info,p))) 
-	{
-		if(p->topic_id==topic_id)
-		{
-			struct stellar_mq_subscriber *tmp_subscriber=topic->subscribers;
-			int cnt=0;
-			while(tmp_subscriber)
-			{
-				if(cnt==p->subscriber_idx)
-				{
-					tmp_subscriber->pkt_msg_cb=plugin_on_msg_cb;
-					return 0;
-				}
-				cnt++;
-				tmp_subscriber=tmp_subscriber->next;
-			}
-		}
-	};
-
-	struct stellar_mq_subscriber *new_subscriber = CALLOC(struct stellar_mq_subscriber,1);
-	new_subscriber->topic_subscriber_idx = topic->subscriber_cnt;
-	new_subscriber->plugin_idx = plugin_idx;
-    new_subscriber->pkt_msg_cb = plugin_on_msg_cb;
-	DL_APPEND(topic->subscribers, new_subscriber);
-
-	struct stellar_mq_subscriber_info sub_info;
-	memset(&sub_info, 0, sizeof(struct stellar_mq_subscriber_info));
-	sub_info.topic_id=topic_id;
-	sub_info.subscriber_idx=topic->subscriber_cnt;	
-	utarray_push_back(packet_plugin_schema->registed_packet_mq_subscriber_info, &sub_info);
-	topic->subscriber_cnt+=1;
-	plug_mgr->packet_topic_subscriber_num+=1;
-	return 0;	
-}
-
-int packet_mq_publish_message(struct packet *pkt, int topic_id, void *msg)
-{
-	struct stellar *st = packet_stellar_get(pkt);
-	assert(st);
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	assert(plug_mgr);
-	int tid = stellar_get_current_thread_id(st);
-	if(stellar_current_thread_packet_mq_counter_get(plug_mgr) >= plug_mgr->max_message_dispatch)return -1;//packet mq donot contain intrinisic msg
-	int ret=stellar_mq_publish_message(topic_id, msg, plug_mgr->packet_mq_schema_array, &(plug_mgr->per_thread_data[tid].per_thread_pkt_mq.mq));
-	if(ret==0)stellar_current_thread_packet_mq_counter_inc(plug_mgr);
-	return ret;
-}
-
-// TODO: limit maximum pub message number in one loop 
-static void plugin_manager_packet_message_dispatch(struct packet *pkt)
-{
-
-	struct stellar *st = packet_stellar_get(pkt);
-	assert(st);
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	assert(plug_mgr);
-	
-	if(plug_mgr->packet_mq_schema_array==NULL)return;
-
-	int tid = stellar_get_current_thread_id(st);
-
-	struct stellar_message **mq= &(plug_mgr->per_thread_data[tid].per_thread_pkt_mq.mq);
-
-	struct stellar_message *mq_elt=NULL, *mq_tmp=NULL;
-	struct stellar_mq_subscriber *sub_elt, *sub_tmp;
-	struct stellar_mq_topic_schema *topic;
-	struct registered_packet_plugin_schema *packet_plugin_schema;
-    while (*mq != NULL)
-    {
-        DL_FOREACH_SAFE(*mq, mq_elt, mq_tmp)
-        {
-            topic = (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->packet_mq_schema_array,
-                                                                    (unsigned int)(mq_elt->topic_id));
-            if (topic)
-            {
-                DL_FOREACH_SAFE(topic->subscribers, sub_elt, sub_tmp)
-                {
-                    if (sub_elt->pkt_msg_cb)
-					{
-						packet_plugin_schema = (struct registered_packet_plugin_schema *)utarray_eltptr(plug_mgr->registered_packet_plugin_array, (unsigned int)sub_elt->plugin_idx);
-						if(packet_plugin_schema)
-						{
-								sub_elt->pkt_msg_cb(pkt, mq_elt->topic_id, mq_elt->msg_data, packet_plugin_schema->plugin_env);
-						}
-					}
-                }
-				if (topic->pkt_msg_free_cb)
-                {
-                    topic->pkt_msg_free_cb(pkt, mq_elt->msg_data, topic->free_cb_arg);
-                }
-            }
-            DL_DELETE(*mq, mq_elt);
-			FREE(mq_elt);
-        }
-    }
-	return;
-}
 
 /*******************************
  * SESSION MQ *
@@ -984,7 +708,6 @@ void plugin_manager_on_packet_ingress(struct plugin_manager_schema *plug_mgr, st
 	if(plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
 	struct registered_packet_plugin_schema *p=NULL;
 	unsigned char ip_proto=packet_get_ip_protocol(pkt);
-	stellar_current_thread_packet_mq_counter_reset(plug_mgr);
 	while ((p = (struct registered_packet_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, p)))
 	{
 		if(p->ip_protocol == ip_proto && p->on_packet)
@@ -992,17 +715,9 @@ void plugin_manager_on_packet_ingress(struct plugin_manager_schema *plug_mgr, st
 			p->on_packet(pkt, ip_proto, p->plugin_env);
 		}
 	}
-	plugin_manager_packet_message_dispatch(pkt);
 	return;
 }
 
-void plugin_manager_on_packet_egress(struct plugin_manager_schema *plug_mgr, struct packet *pkt)
-{
-	if(plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
-	plugin_manager_packet_message_dispatch(pkt);
-	per_thread_packet_exdata_arrary_clean(plug_mgr, pkt);
-}
-
 /*********************************************
  * PLUGIN MANAGER POLLING PLUGIN *
  *********************************************/
diff --git a/src/plugin_manager/plugin_manager.h b/src/plugin_manager/plugin_manager.h
index 6810da8..2d07c25 100644
--- a/src/plugin_manager/plugin_manager.h
+++ b/src/plugin_manager/plugin_manager.h
@@ -9,7 +9,6 @@ struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char
 void plugin_manager_exit(struct plugin_manager_schema *plug_mgr);
 
 void plugin_manager_on_packet_ingress(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
-void plugin_manager_on_packet_egress(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
 
 //return polling work state, 0: idle, 1: working
 int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr);
diff --git a/src/plugin_manager/plugin_manager_interna.h b/src/plugin_manager/plugin_manager_interna.h
index 0cf55d6..6b2beba 100644
--- a/src/plugin_manager/plugin_manager_interna.h
+++ b/src/plugin_manager/plugin_manager_interna.h
@@ -4,44 +4,22 @@
 
 #include "stellar/session_exdata.h"
 #include "stellar/session_mq.h"
-#include "stellar/packet_exdata.h"
-#include "stellar/packet_mq.h"
 
 #include "bitmap/bitmap.h"
 #include "uthash/utarray.h"
 
-struct per_thread_exdata_array
-{
-	struct stellar_exdata *exdata_array;
-};
-
 struct stellar_message;
-struct per_thread_mq
-{
-	struct stellar_message *mq;
-	int pub_msg_cnt;
-};
-
-struct plugin_manger_per_thread_data
-{
-	struct per_thread_exdata_array per_thread_pkt_exdata_array;
-	struct per_thread_mq per_thread_pkt_mq;
-};
 
 struct plugin_manager_schema
 {
 	struct stellar *st;
 	UT_array *plugin_load_specs_array;
-	UT_array *packet_exdata_schema_array;
-	UT_array *packet_mq_schema_array;
 	UT_array *session_exdata_schema_array;
 	UT_array *session_mq_schema_array;
 	UT_array *registered_session_plugin_array;
 	UT_array *registered_packet_plugin_array;
 	UT_array *registered_polling_plugin_array;
-	int packet_mq_topic_num;
 	int session_mq_topic_num;
-	int packet_topic_subscriber_num;
 	int session_topic_subscriber_num;
 	int tcp_topic_id;
 	int tcp_stream_topic_id;
@@ -49,12 +27,9 @@ struct plugin_manager_schema
 	int egress_topic_id;
 	int control_packet_topic_id;
 	int max_message_dispatch;
-	struct plugin_manger_per_thread_data *per_thread_data;
 }__attribute__((aligned(sizeof(void*))));
 
 
-
-
 struct stellar_exdata
 {
 	void *exdata;
@@ -69,7 +44,6 @@ struct stellar_exdata_schema
 	{
 		void *free_func;
 		session_exdata_free *sess_free_func;
-		packet_exdata_free *pkt_free_func;
 	};
 	
 	void *free_arg;
@@ -91,7 +65,6 @@ typedef struct stellar_mq_subscriber
 	union 
 	{
     	on_session_msg_cb_func *sess_msg_cb;
-    	on_packet_msg_cb_func *pkt_msg_cb;
 	};
     struct stellar_mq_subscriber *next, *prev;
 }stellar_mq_subscriber __attribute__((aligned(sizeof(void*))));
@@ -108,7 +81,6 @@ struct stellar_mq_topic_schema
 	{
 		void *free_cb;
 		session_msg_free_cb_func *sess_msg_free_cb;
-		packet_msg_free_cb_func *pkt_msg_free_cb;
 	};
 	struct stellar_mq_subscriber *subscribers; 
 }__attribute__((aligned(sizeof(void*))));
diff --git a/src/stellar_on_sapp/stellar_on_sapp_api.c b/src/stellar_on_sapp/stellar_on_sapp_api.c
index f7dd17d..5974ace 100644
--- a/src/stellar_on_sapp/stellar_on_sapp_api.c
+++ b/src/stellar_on_sapp/stellar_on_sapp_api.c
@@ -60,7 +60,6 @@ inline struct plugin_manager_runtime * session_plugin_manager_runtime_get(struct
 /*********************************************
  * STELLAR INIT & EXIT ON SAPP *
  *********************************************/
-
 struct stellar *stellar_init_on_sapp(const char *toml_conf_path)
 {
 	struct stellar *st = CALLOC(struct stellar, 1);
@@ -157,7 +156,6 @@ void session_defer_on_sapp(struct session *sess)
 		if(pkt->raw_pkt)
 		{
 			plugin_manager_on_session_egress(sess, pkt);
-			plugin_manager_on_packet_egress(sess->st->plug_mgr, pkt);
 		}
 	}
 	sess->cur_pkt.raw_pkt=NULL;//clear raw_pkt
@@ -193,10 +191,6 @@ void packet_update_on_sapp(struct stellar *st, struct streaminfo *pstream, void
 	}
 	//FIXME: defer TCP/UDP packet on session update
 	plugin_manager_on_packet_ingress(st->plug_mgr, &pkt);
-	if(pkt.ip_proto!=IPPROTO_TCP && pkt.ip_proto!=IPPROTO_UDP)
-	{
-		plugin_manager_on_packet_egress(st->plug_mgr, &pkt);
-	}
 }
 
 inline int polling_on_sapp(struct stellar *st)
diff --git a/test/plugin_manager/plugin_manager_gtest_main.cpp b/test/plugin_manager/plugin_manager_gtest_main.cpp
index 75f928c..aa1a359 100644
--- a/test/plugin_manager/plugin_manager_gtest_main.cpp
+++ b/test/plugin_manager/plugin_manager_gtest_main.cpp
@@ -1,5 +1,4 @@
 #include <gtest/gtest.h>
-#include <limits.h>
 
 #include "stellar/utils.h"
 #include "plugin_manager_gtest_mock.h"
@@ -16,8 +15,6 @@ void whitebox_test_plugin_manager_intrisic_metadata(struct stellar *st, struct p
 	EXPECT_TRUE(plug_mgr->plugin_load_specs_array==NULL);
 
 	//packet exdata & mq schema null
-	EXPECT_TRUE(plug_mgr->packet_exdata_schema_array==NULL);
-	EXPECT_TRUE(plug_mgr->packet_mq_schema_array==NULL);
 
 	//session exdata schema null
 	EXPECT_TRUE(plug_mgr->session_exdata_schema_array==NULL);
@@ -30,8 +27,6 @@ void whitebox_test_plugin_manager_intrisic_metadata(struct stellar *st, struct p
 	EXPECT_TRUE(plug_mgr->registered_packet_plugin_array==NULL);
 	EXPECT_TRUE(plug_mgr->registered_session_plugin_array==NULL);
 
-	EXPECT_EQ(plug_mgr->packet_mq_topic_num, 0);
-
 	int intrinsic_topic_num=utarray_len(plug_mgr->session_mq_schema_array);
 	EXPECT_EQ(plug_mgr->session_mq_topic_num, intrinsic_topic_num);//TCP,UDP,TCP_STREAM,EGRESS,CONTROL
 
@@ -50,14 +45,6 @@ void whitebox_test_plugin_manager_intrisic_metadata(struct stellar *st, struct p
 	topic = (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->session_mq_schema_array, (unsigned int)plug_mgr->control_packet_topic_id);
 	EXPECT_STREQ(topic->topic_name, TOPIC_CONTROL_PACKET);
 
-	EXPECT_TRUE(plug_mgr->per_thread_data!=NULL);
-	int thread_num=stellar_get_worker_thread_num(st);
-	for(int i=0; i<thread_num; i++)
-	{
-		EXPECT_TRUE(plug_mgr->per_thread_data[i].per_thread_pkt_exdata_array.exdata_array==NULL);
-		EXPECT_TRUE(plug_mgr->per_thread_data[i].per_thread_pkt_mq.mq==NULL);
-	}
-
 	//intrinsic topic
 	EXPECT_GE(stellar_session_mq_get_topic_id(st, TOPIC_TCP), 0);
 	EXPECT_GE(stellar_session_mq_get_topic_id(st, TOPIC_TCP_STREAM), 0);
@@ -80,219 +67,25 @@ TEST(plugin_manager_init, init_with_null_toml) {
 	plugin_manager_exit(plug_mgr);
 }
 
-/******************************************
- * TEST PLUGIN MANAGER PACKET PLUGIN INIT *
- ******************************************/
-
-static void test_mock_packet_exdata_free(struct packet *pkt, int idx, void *ex_ptr, void *arg){}
-
-static void test_mock_overwrite_packet_exdata_free(struct packet *pkt, int idx, void *ex_ptr, void *arg){}
-
-TEST(plugin_manager_init, packet_exdata_new_index_overwrite) {
-	struct stellar st={0};
-	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
-	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
-
-	const char *exdata_name="PACKET_EXDATA";
-	int exdata_idx=stellar_packet_exdata_new_index(&st,exdata_name, test_mock_packet_exdata_free, &st);
-	EXPECT_GE(exdata_idx, 0);
-	int overwrite_idx=stellar_packet_exdata_new_index(&st,exdata_name, test_mock_overwrite_packet_exdata_free, plug_mgr);
-	EXPECT_GE(overwrite_idx, 0);
-	EXPECT_EQ(overwrite_idx, exdata_idx);
-
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        struct stellar_exdata_schema *exdata_schema = (struct stellar_exdata_schema *)utarray_eltptr(
-            plug_mgr->packet_exdata_schema_array, (unsigned int)exdata_idx);
-        EXPECT_EQ(exdata_schema->free_func, (void *)test_mock_overwrite_packet_exdata_free);
-        EXPECT_EQ(exdata_schema->free_arg, plug_mgr);
-        EXPECT_EQ(exdata_schema->idx, exdata_idx);
-        EXPECT_STREQ(exdata_schema->name, exdata_name);
-
-        int exdata_num = utarray_len(plug_mgr->packet_exdata_schema_array);
-        EXPECT_EQ(exdata_num, 1);
-    }
-
-    plugin_manager_exit(plug_mgr);
-}
-
-void test_mock_packet_msg_free(struct packet *pkt, void *msg, void *msg_free_arg){}
-void test_mock_overwrite_packet_msg_free(struct packet *pkt, void *msg, void *msg_free_arg){}
-
-TEST(plugin_manager_init, packet_mq_topic_create_and_update) {
-	struct stellar st={0};
-	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
-	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
-
-	const char *topic_name="PACKET_TOPIC";
-
-    EXPECT_EQ(stellar_packet_mq_get_topic_id(&st, topic_name), -1); // illegal topic_name
-
-    int topic_id = stellar_packet_mq_create_topic(&st, topic_name, test_mock_packet_msg_free, &st);
-    EXPECT_GE(topic_id, 0);
-    struct stellar_mq_topic_schema *topic_schema = NULL;
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        topic_schema =
-            (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->packet_mq_schema_array, (unsigned int)topic_id);
-        EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_packet_msg_free);
-        EXPECT_EQ(topic_schema->free_cb_arg, &st);
-        EXPECT_EQ(topic_schema->topic_id, topic_id);
-        EXPECT_STREQ(topic_schema->topic_name, topic_name);
-    }
-
-    EXPECT_EQ(stellar_packet_mq_get_topic_id(&st, topic_name), topic_id);
-    EXPECT_EQ(stellar_packet_mq_create_topic(&st, topic_name, test_mock_overwrite_packet_msg_free, plug_mgr),
-              -1); // duplicate create, return error
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        topic_schema =
-            (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->packet_mq_schema_array, (unsigned int)topic_id);
-        EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_packet_msg_free);
-        EXPECT_EQ(topic_schema->free_cb_arg, &st);
-        EXPECT_EQ(topic_schema->topic_id, topic_id);
-        EXPECT_STREQ(topic_schema->topic_name, topic_name);
-    }
-
-    EXPECT_EQ(stellar_packet_mq_update_topic(&st, topic_id, test_mock_overwrite_packet_msg_free, plug_mgr), 0);
-
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        topic_schema =
-            (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->packet_mq_schema_array, (unsigned int)topic_id);
-        EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_overwrite_packet_msg_free);
-        EXPECT_EQ(topic_schema->free_cb_arg, plug_mgr);
-        EXPECT_EQ(topic_schema->topic_id, topic_id);
-        EXPECT_STREQ(topic_schema->topic_name, topic_name);
-        EXPECT_EQ(utarray_len(plug_mgr->packet_mq_schema_array), 1);
-    }
-
-    EXPECT_EQ(stellar_packet_mq_destroy_topic(&st, 10), -1); // illgeal topic_id
-
-    EXPECT_EQ(stellar_packet_mq_destroy_topic(&st, topic_id), 1);
-    EXPECT_EQ(stellar_packet_mq_destroy_topic(&st, topic_id), 0); // duplicate destroy, return 0;
-
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        EXPECT_EQ(utarray_len(plug_mgr->packet_mq_schema_array), 1); // destory won't delete the topic schema
-        EXPECT_EQ(plug_mgr->packet_mq_topic_num, 0);
-    }
-    plugin_manager_exit(plug_mgr);
-}
-
-void test_mock_on_packet_msg(struct packet *pkt, int topic_id, const void *msg, void *plugin_env){}
-
-void test_mock_overwrite_on_packet_msg(struct packet *pkt, int topic_id, const void *msg, void *plugin_env){}
-
-TEST(plugin_manager_init, packet_mq_subscribe) {
-
-	struct stellar st={0};
-	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
-	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
-
-
-	const char *topic_name="PACKET_TOPIC";
-
-	int topic_id=stellar_packet_mq_create_topic(&st, topic_name, test_mock_packet_msg_free, &st);
-	EXPECT_GE(topic_id, 0);
-
-	EXPECT_EQ(stellar_packet_mq_subscribe(&st, topic_id, test_mock_on_packet_msg, 10+PACKET_PULGIN_ID_BASE),-1);//illgeal plugin_id
-	EXPECT_EQ(stellar_packet_mq_subscribe(&st, 10, test_mock_on_packet_msg, 10+PACKET_PULGIN_ID_BASE),-1);//illgeal topic_id & plugin_id
-
-	int plugin_id=stellar_packet_plugin_register(&st, 6, NULL, &st);
-	EXPECT_GE(plugin_id, PACKET_PULGIN_ID_BASE);
-
-	EXPECT_EQ(stellar_packet_mq_subscribe(&st, topic_id, test_mock_on_packet_msg, plugin_id),0);
-	EXPECT_EQ(stellar_packet_mq_subscribe(&st, topic_id, test_mock_overwrite_on_packet_msg, plugin_id),0);//duplicate subscribe, return 0, won't overwrite
-	struct stellar_mq_topic_schema *topic_schema;
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        topic_schema = (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->packet_mq_schema_array, (unsigned int)topic_id);
-        EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_packet_msg_free);
-        EXPECT_EQ(topic_schema->free_cb_arg, &st);
-        EXPECT_EQ(topic_schema->topic_id, topic_id);
-        EXPECT_STREQ(topic_schema->topic_name, topic_name);
-    }
-
-    EXPECT_EQ(topic_schema->subscriber_cnt, 1);
-	EXPECT_EQ(topic_schema->subscribers->pkt_msg_cb, (void *)test_mock_overwrite_on_packet_msg);
-
-	plugin_manager_exit(plug_mgr);
-}
-
 /*******************************************
  * TEST PLUGIN MANAGER PACKET PLUGIN RUNTIME*
  *******************************************/
 
 #define PACKET_PROTO_PLUGIN_NUM 128
-#define PACKET_EXDATA_NUM 2
-#define PACKET_TOPIC_NUM 2
-#define PACKET_MQ_SUB_NUM 2
 struct packet_plugin_env
 {
 	struct plugin_manager_schema *plug_mgr;
 	int basic_on_packet_called;
 	int proto_filter_plugin_id[PACKET_PROTO_PLUGIN_NUM];
 	int proto_filter_plugin_called[PACKET_PROTO_PLUGIN_NUM];
-	int exdata_set_on_packet_called;
-	int exdata_get_on_packet_called;
-	unsigned int packet_exdata_idx[PACKET_EXDATA_NUM];
-	int exdata_free_called[PACKET_EXDATA_NUM];
-	unsigned int packet_topic_id[PACKET_TOPIC_NUM];
-	unsigned int packet_mq_sub_plugin_id[PACKET_MQ_SUB_NUM];
-	int msg_pub_cnt;
-	int msg_sub_cnt;
-	int msg_free_cnt;
 };
 
-static void test_basic_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
-	EXPECT_TRUE(env!=NULL);
-	EXPECT_EQ(pkt->ip_proto, ip_protocol);
-	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-	EXPECT_EQ(packet_exdata_set(pkt, 2, pkt), -1);// illegal set
-	EXPECT_EQ(packet_exdata_get(pkt, 2), nullptr);// illegal get
-	env->basic_on_packet_called+=1;
-	return;
-}
-
-TEST(plugin_manager, packet_plugin_illegal_exdata) {
-
-	struct stellar st={0};
-	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
-	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
-
-	unsigned char ip_proto=6;
-	struct packet_plugin_env env;
-	memset(&env, 0, sizeof(struct packet_plugin_env));
-	env.plug_mgr=plug_mgr;
-	int plugin_id=stellar_packet_plugin_register(&st, ip_proto, test_basic_on_packet, &env);
-	EXPECT_GE(plugin_id, PACKET_PULGIN_ID_BASE);
-
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        int packet_plugin_num = utarray_len(plug_mgr->registered_packet_plugin_array);
-        EXPECT_EQ(packet_plugin_num, 1);
-    }
-
-    struct packet pkt={&st, IPv4, ip_proto};
-	plugin_manager_on_packet_ingress(plug_mgr, &pkt);
-	plugin_manager_on_packet_egress(plug_mgr, &pkt);
-	
-	plugin_manager_exit(plug_mgr);
-
-	EXPECT_EQ(env.basic_on_packet_called, 1);
-}
-
 static void test_proto_filter_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
 {
 	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
 	EXPECT_TRUE(env!=NULL);
 	EXPECT_EQ(pkt->ip_proto, ip_protocol);
 	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-	EXPECT_EQ(packet_exdata_set(pkt, 2, pkt), -1);// illegal set
-	EXPECT_EQ(packet_exdata_get(pkt, 2), nullptr);// illegal get
 	env->proto_filter_plugin_called[ip_protocol]+=1;
 	return;
 }
@@ -330,7 +123,6 @@ TEST(plugin_manager, packet_plugins_with_proto_filter) {
         {
             pkt.ip_proto = i;
             plugin_manager_on_packet_ingress(plug_mgr, &pkt);
-            plugin_manager_on_packet_egress(plug_mgr, &pkt);
         }
     }
     plugin_manager_exit(plug_mgr);
@@ -341,339 +133,6 @@ TEST(plugin_manager, packet_plugins_with_proto_filter) {
 	}
 }
 
-static void test_exdata_set_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
-	EXPECT_TRUE(env!=NULL);
-	EXPECT_EQ(pkt->ip_proto, ip_protocol);
-	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-	env->exdata_set_on_packet_called+=1;
-
-	int exdata_idx_len=(int)(sizeof(env->packet_exdata_idx) / sizeof(env->packet_exdata_idx[0]));
-
-	for(int i=0; i<exdata_idx_len; i++)
-	{
-		long long *exdata_val=CALLOC(long long , 1);
-		*exdata_val=i;
-		EXPECT_EQ(packet_exdata_set(pkt, env->packet_exdata_idx[i], exdata_val), 0);
-	}
-	return;
-}
-
-static void test_exdata_get_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
-	EXPECT_TRUE(env!=NULL);
-	EXPECT_EQ(pkt->ip_proto, ip_protocol);
-	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-
-	int exdata_idx_len=(int)(sizeof(env->packet_exdata_idx) / sizeof(env->packet_exdata_idx[0]));
-	
-	for(int i=0; i<exdata_idx_len; i++)
-	{
-		long long *exdata_val=(long long *)packet_exdata_get(pkt, env->packet_exdata_idx[i]);
-		EXPECT_EQ(*exdata_val, i);
-	}
-	env->exdata_get_on_packet_called+=1;
-	return;
-}
-
-static void test_packet_exdata_free(struct packet *pkt, int idx, void *ex_ptr, void *arg)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)arg;
-	EXPECT_EQ(env->packet_exdata_idx[idx], idx);
-	EXPECT_EQ(*(long long *)ex_ptr, idx);
-	FREE(ex_ptr);
-	env->exdata_free_called[idx]+=1;
-	return;
-}
-
-
-TEST(plugin_manager, packet_plugins_share_exdata) {
-
-	struct stellar st={0};
-	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
-	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
-
-	unsigned char ip_proto=6;
-	struct packet_plugin_env env;
-	memset(&env, 0, sizeof(struct packet_plugin_env));
-	env.plug_mgr=plug_mgr;
-
-	char exdata_name[PACKET_EXDATA_NUM][NAME_MAX];
-	int exdata_idx_len=(int)(sizeof(env.packet_exdata_idx) / sizeof(env.packet_exdata_idx[0]));
-	for(int i=0; i<exdata_idx_len; i++)
-	{
-		sprintf(exdata_name[i], "PACKET_EXDATA_%d", i);
-		env.packet_exdata_idx[i]=stellar_packet_exdata_new_index(&st, exdata_name[i], test_packet_exdata_free, &env);
-        {
-            SCOPED_TRACE("White-box test, check stellar internal schema");
-            struct stellar_exdata_schema *exdata_schema = (struct stellar_exdata_schema *)utarray_eltptr(
-                plug_mgr->packet_exdata_schema_array, env.packet_exdata_idx[i]);
-
-            EXPECT_EQ(exdata_schema->free_func, (void *)test_packet_exdata_free);
-            EXPECT_EQ(exdata_schema->free_arg, &env);
-            EXPECT_EQ(exdata_schema->idx, env.packet_exdata_idx[i]);
-            EXPECT_STREQ(exdata_schema->name, exdata_name[i]);
-        }
-    }
-
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        EXPECT_EQ(utarray_len(plug_mgr->packet_exdata_schema_array), exdata_idx_len);
-    }
-
-    int exdata_set_plugin_id=stellar_packet_plugin_register(&st, ip_proto, test_exdata_set_on_packet, &env);
-	EXPECT_GE(exdata_set_plugin_id, PACKET_PULGIN_ID_BASE);
-
-	int exdata_get_plugin_id=stellar_packet_plugin_register(&st, ip_proto, test_exdata_get_on_packet, &env);
-	EXPECT_GE(exdata_get_plugin_id, PACKET_PULGIN_ID_BASE);
-
-    {
-        SCOPED_TRACE("White-box test, check stellar internal schema");
-        EXPECT_EQ(utarray_len(plug_mgr->registered_packet_plugin_array), 2); // Fix plugin number
-    }
-
-    struct packet pkt={&st, IPv4, ip_proto};
-
-	int N_packet=10;
-
-	for(int i=0; i < N_packet; i++)
-	{
-		plugin_manager_on_packet_ingress(plug_mgr, &pkt);
-		plugin_manager_on_packet_egress(plug_mgr, &pkt);
-	}
-	plugin_manager_exit(plug_mgr);
-
-	EXPECT_EQ(env.exdata_set_on_packet_called, N_packet);
-	EXPECT_EQ(env.exdata_get_on_packet_called, N_packet);
-
-	for(int i=0; i < exdata_idx_len; i++)
-	{
-		EXPECT_EQ(env.exdata_free_called[i], N_packet);
-	}
-}
-
-static void test_packet_msg_free_cb_func(struct packet *pkt, void *msg, void *msg_free_arg)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)msg_free_arg;
-	EXPECT_EQ(pkt, msg);
-	env->msg_free_cnt+=1;
-	return;
-}
-
-static void test_mq_on_packet_msg(struct packet *pkt, int topic_id, const void *msg, void *plugin_env)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
-	EXPECT_TRUE(env!=NULL);
-	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-	env->msg_sub_cnt+=1;
-	return;
-}
-
-static void test_mq_pub_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
-	EXPECT_TRUE(env!=NULL);
-	EXPECT_EQ(pkt->ip_proto, ip_protocol);
-	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-	int topic_id_num=(int)(sizeof(env->packet_topic_id) / sizeof(env->packet_topic_id[0]));
-	for(int i=0; i<topic_id_num; i++)
-	{
-		EXPECT_EQ(packet_mq_publish_message(pkt, env->packet_topic_id[i], pkt), 0);
-		env->msg_pub_cnt+=1;
-	}
-	return;
-}
-
-TEST(plugin_manager, packet_plugins_mq_pub_sub) {
-
-	struct stellar st={0};
-	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
-	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
-
-	unsigned char ip_proto=6;
-	struct packet_plugin_env env;
-	memset(&env, 0, sizeof(struct packet_plugin_env));
-	env.plug_mgr=plug_mgr;
-	char topic_name[PACKET_TOPIC_NUM][NAME_MAX];
-
-	int topic_id_num=(int)(sizeof(env.packet_topic_id) / sizeof(env.packet_topic_id[0]));
-
-	for(int i=0; i<topic_id_num; i++)
-	{
-		sprintf(topic_name[i], "PACKET_TOPIC_%d", i);
-		env.packet_topic_id[i]=stellar_packet_mq_create_topic(&st, topic_name[i], test_packet_msg_free_cb_func, &env);
-		EXPECT_GE(env.packet_topic_id[i], 0);
-        {
-            SCOPED_TRACE("White-box test, check stellar internal schema");
-            struct stellar_mq_topic_schema *topic = (struct stellar_mq_topic_schema *)utarray_eltptr(
-                plug_mgr->packet_mq_schema_array, env.packet_topic_id[i]);
-            EXPECT_EQ(topic->free_cb, test_packet_msg_free_cb_func);
-            EXPECT_EQ(topic->free_cb_arg, &env);
-            EXPECT_EQ(topic->topic_id, env.packet_topic_id[i]);
-            EXPECT_STREQ(topic->topic_name, topic_name[i]);
-        }
-    }
-
-	{
-		SCOPED_TRACE("White-box test, check stellar internal schema");
-		EXPECT_EQ(utarray_len(plug_mgr->packet_mq_schema_array), topic_id_num);
-	}
-
-	int pub_plugin_id=stellar_packet_plugin_register(&st, ip_proto, test_mq_pub_on_packet, &env);
-	EXPECT_GE(pub_plugin_id, PACKET_PULGIN_ID_BASE);
-
-	int topic_sub_num=(int)(sizeof(env.packet_mq_sub_plugin_id) / sizeof(env.packet_mq_sub_plugin_id[0]));
-
-    for (int i = 0; i < topic_sub_num; i++)
-    {
-        env.packet_mq_sub_plugin_id[i] = stellar_packet_plugin_register(&st, ip_proto, NULL, &env);// empty on_packet is ok
-        EXPECT_GE(env.packet_mq_sub_plugin_id[i], PACKET_PULGIN_ID_BASE);
-		for(int j = 0; j < topic_id_num; j++)
-		{
-			EXPECT_EQ(stellar_packet_mq_subscribe(&st, env.packet_topic_id[j], test_mq_on_packet_msg, env.packet_mq_sub_plugin_id[i]), 0);
-		}
-    }
-
-	{
-		SCOPED_TRACE("White-box test, check stellar internal schema");
-		EXPECT_EQ(utarray_len(plug_mgr->registered_packet_plugin_array), topic_sub_num+1);
-	}
-
-	struct packet pkt={&st, IPv4, ip_proto};
-
-	int N_packet=10;
-    for (int i = 0; i < N_packet; i++)
-    {
-        plugin_manager_on_packet_ingress(plug_mgr, &pkt);
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
-    }
-
-    plugin_manager_exit(plug_mgr);
-	EXPECT_EQ(N_packet*topic_id_num, env.msg_pub_cnt);
-	EXPECT_EQ(env.msg_free_cnt, env.msg_pub_cnt);
-	EXPECT_EQ(env.msg_sub_cnt, env.msg_pub_cnt*topic_sub_num);
-}
-
-static void overlimit_packet_msg_free_cb_func(struct packet *pkt, void *msg, void *msg_free_arg)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)msg_free_arg;
-	env->msg_free_cnt+=1;
-	FREE(msg);
-	return;
-}
-
-static void overlimit_sub_on_packet_msg(struct packet *pkt, int topic_id, const void *msg, void *plugin_env)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
-	EXPECT_TRUE(env!=NULL);
-	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-	env->msg_sub_cnt+=1;
-	return;
-}
-
-static void overlimit_pub_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
-{
-	struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
-	EXPECT_TRUE(env!=NULL);
-	EXPECT_EQ(pkt->ip_proto, ip_protocol);
-	EXPECT_EQ(pkt->st, env->plug_mgr->st);
-	int topic_id_num=(int)(sizeof(env->packet_topic_id) / sizeof(env->packet_topic_id[0]));
-	int cnt=0;
-	int *msg;
-	for(int i=0; i<topic_id_num; i++)
-	{
-		for(int j=0; j < MAX_MSG_PER_DISPATCH; j++)
-		{
-			msg=CALLOC(int, 1);
-			*msg=cnt;
-			int pub_ret=packet_mq_publish_message(pkt, env->packet_topic_id[i], msg);
-			if(cnt < MAX_MSG_PER_DISPATCH)
-			{
-				ASSERT_EQ(pub_ret, 0);
-				env->msg_pub_cnt+=1;
-			}
-			else
-			{
-				ASSERT_EQ(pub_ret, -1);
-			}
-			if(pub_ret!=0)FREE(msg);
-			cnt+=1;
-		}
-	}
-	return;
-}
-
-TEST(plugin_manager, packet_plugins_pub_overlimit) {
-
-	struct stellar st={0};
-	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
-	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
-
-	unsigned char ip_proto=6;
-	struct packet_plugin_env env;
-	memset(&env, 0, sizeof(struct packet_plugin_env));
-	env.plug_mgr=plug_mgr;
-	char topic_name[PACKET_TOPIC_NUM][NAME_MAX];
-
-	int topic_id_num=(int)(sizeof(env.packet_topic_id) / sizeof(env.packet_topic_id[0]));
-
-	for(int i=0; i<topic_id_num; i++)
-	{
-		sprintf(topic_name[i], "PACKET_TOPIC_%d", i);
-		env.packet_topic_id[i]=stellar_packet_mq_create_topic(&st, topic_name[i], overlimit_packet_msg_free_cb_func, &env);
-		EXPECT_GE(env.packet_topic_id[i], 0);
-        {
-            SCOPED_TRACE("White-box test, check stellar internal schema");
-            struct stellar_mq_topic_schema *topic = (struct stellar_mq_topic_schema *)utarray_eltptr(
-                plug_mgr->packet_mq_schema_array, env.packet_topic_id[i]);
-            EXPECT_EQ(topic->free_cb, overlimit_packet_msg_free_cb_func);
-            EXPECT_EQ(topic->free_cb_arg, &env);
-            EXPECT_EQ(topic->topic_id, env.packet_topic_id[i]);
-            EXPECT_STREQ(topic->topic_name, topic_name[i]);
-        }
-    }
-
-	{
-		SCOPED_TRACE("White-box test, check stellar internal schema");
-		EXPECT_EQ(utarray_len(plug_mgr->packet_mq_schema_array), topic_id_num);
-	}
-
-	int pub_plugin_id=stellar_packet_plugin_register(&st, ip_proto, overlimit_pub_on_packet, &env);
-	EXPECT_GE(pub_plugin_id, PACKET_PULGIN_ID_BASE);
-
-	int topic_sub_num=(int)(sizeof(env.packet_mq_sub_plugin_id) / sizeof(env.packet_mq_sub_plugin_id[0]));
-
-    for (int i = 0; i < topic_sub_num; i++)
-    {
-        env.packet_mq_sub_plugin_id[i] = stellar_packet_plugin_register(&st, ip_proto, NULL, &env);// empty on_packet is ok
-        EXPECT_GE(env.packet_mq_sub_plugin_id[i], PACKET_PULGIN_ID_BASE);
-		for(int j = 0; j < topic_id_num; j++)
-		{
-			EXPECT_EQ(stellar_packet_mq_subscribe(&st, env.packet_topic_id[j], overlimit_sub_on_packet_msg, env.packet_mq_sub_plugin_id[i]), 0);
-		}
-    }
-
-	{
-		SCOPED_TRACE("White-box test, check stellar internal schema");
-		EXPECT_EQ(utarray_len(plug_mgr->registered_packet_plugin_array), topic_sub_num+1);
-	}
-
-	struct packet pkt={&st, IPv4, ip_proto};
-
-	int N_packet=10;
-    for (int i = 0; i < N_packet; i++)
-    {
-        plugin_manager_on_packet_ingress(plug_mgr, &pkt);
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
-    }
-
-    plugin_manager_exit(plug_mgr);
-	EXPECT_EQ(N_packet*MAX_MSG_PER_DISPATCH, env.msg_pub_cnt);
-	EXPECT_EQ(env.msg_free_cnt, env.msg_pub_cnt);
-	EXPECT_EQ(env.msg_sub_cnt, env.msg_pub_cnt*topic_sub_num);
-}
 /**********************************************
  * TEST PLUGIN MANAGER ON SESSION PLUGIN INIT *
  **********************************************/
@@ -760,7 +219,6 @@ TEST(plugin_manager_init, session_mq_topic_create_and_update) {
         EXPECT_EQ(utarray_len(plug_mgr->session_mq_schema_array), 6); // 5 intrinsic topic + 1 created topic
     }
 
-    EXPECT_EQ(stellar_packet_mq_destroy_topic(&st, 10), -1);// illgeal packet topic_id
 	EXPECT_EQ(stellar_session_mq_destroy_topic(&st, 10), -1);// illgeal session topic_id
 
 	EXPECT_EQ(stellar_session_mq_destroy_topic(&st, topic_id), 1);
@@ -810,7 +268,7 @@ TEST(plugin_manager_init, session_mq_subscribe_overwrite) {
 	}
 
 	EXPECT_EQ(topic_schema->subscriber_cnt, 1);
-	EXPECT_EQ(topic_schema->subscribers->pkt_msg_cb, (void *)test_mock_overwrite_on_session_msg);
+	EXPECT_EQ(topic_schema->subscribers->sess_msg_cb, (void *)test_mock_overwrite_on_session_msg);
 
 	plugin_manager_exit(plug_mgr);
 }
@@ -877,7 +335,6 @@ TEST(plugin_manager, no_plugin_register_runtime) {
             plugin_manager_on_session_egress(&sess[i], &pkt);
         }
 
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
     }
 
     for(int i=0; i < env.N_session; i++)
@@ -1002,7 +459,6 @@ TEST(plugin_manager, session_plugin_on_intrinsic_ingress_egress) {
             plugin_manager_on_session_egress(&sess[i], &pkt);
         }
 
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
     }
 
     for(int i=0; i < env.N_session; i++)
@@ -1145,7 +601,6 @@ TEST(plugin_manager, session_plugin_ignore_on_ctx_new_sub_other_msg) {
             plugin_manager_on_session_egress(&sess[i], &pkt);
         }
 
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
     }
 
     for(int i=0; i < env.N_session; i++)
@@ -1299,7 +754,6 @@ TEST(plugin_manager, session_plugin_pub_msg_overlimt) {
             plugin_manager_on_session_egress(&sess[i], &pkt);
         }
 
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
     }
 
     for(int i=0; i < env.N_session; i++)
@@ -1381,7 +835,6 @@ TEST(plugin_manager, session_plugin_on_ctx_new_then_dettach) {
             plugin_manager_on_session_egress(&sess[i], &pkt);
         }
 
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
     }
 
     for(int i=0; i < env.N_session; i++)
@@ -1469,7 +922,6 @@ TEST(plugin_manager, session_plugin_pub_on_ctx_free) {
             plugin_manager_on_session_egress(&sess[i], &pkt);
         }
 
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
     }
 
     for(int i=0; i < env.N_session; i++)
@@ -1588,7 +1040,6 @@ TEST(plugin_manager, session_plugin_pub_msg_on_closing) {
             plugin_manager_on_session_egress(&sess[i], &pkt);
         }
 
-        plugin_manager_on_packet_egress(plug_mgr, &pkt);
     }
 
     for(int i=0; i < env.N_session; i++)