✨ feat(packet direction unknonw): set session in session packet

6 files changed, 721 insertions, 166 deletions

diff --git a/include/stellar/session.h b/include/stellar/session.h
index f28e46d..417fc2f 100644
--- a/include/stellar/session.h
+++ b/include/stellar/session.h
@@ -112,6 +112,7 @@ const struct packet *session_get0_current_packet(struct session *sess);
 
 #define PACKET_DIRECTION_C2S 0
 #define PACKET_DIRECTION_S2C 1
+#define PACKET_DIRECTION_UNKNOWN 2
 int packet_get_direction(const struct packet *pkt);
 const char *packet_get0_data(const struct packet *pkt, size_t *data_len);
 int packet_arrive_time(const struct packet *pkt, struct timeval *ts);
diff --git a/src/plugin_manager/plugin_manager.c b/src/plugin_manager/plugin_manager.c
index 7f11fb5..1760c6d 100644
--- a/src/plugin_manager/plugin_manager.c
+++ b/src/plugin_manager/plugin_manager.c
@@ -227,6 +227,8 @@ int stellar_exdata_new_index(struct stellar *st, const char *name, UT_array **ex
 		t_schema = (struct stellar_exdata_schema *)utarray_eltptr(*exdata_schema, i);
 		if(strcmp(t_schema->name, name) == 0)
 		{
+			t_schema->free_func=free_func;
+			t_schema->free_arg=free_arg;
 			return t_schema->idx;
 		}
 	}
@@ -424,7 +426,7 @@ int stellar_mq_create_topic(struct stellar *st, const char *topic_name, void *ms
 
 int stellar_mq_destroy_topic(int topic_id, UT_array *mq_schema_array)
 {
-	if(mq_schema_array==NULL)return 0;
+	if(mq_schema_array==NULL)return -1;
     unsigned int len = utarray_len(mq_schema_array);
     if (len <= (unsigned int)topic_id)
         return -1;
@@ -535,7 +537,6 @@ int stellar_packet_mq_subscribe(struct stellar *st, int topic_id, on_packet_msg_
 				cnt++;
 				tmp_subscriber=tmp_subscriber->next;
 			}
-			return -1;
 		}
 	};
 
@@ -615,11 +616,11 @@ static void plugin_manager_packet_message_dispatch(struct packet *pkt)
  * SESSION MQ *
  *******************************/
 
-void session_mq_free(struct session *sess, struct stellar_message *head, UT_array *mq_schema_array)
+void session_mq_free(struct session *sess, struct stellar_message **head, UT_array *mq_schema_array)
 {
 	struct stellar_message *mq_elt, *tmp;
 	struct stellar_mq_topic_schema *topic;
-    DL_FOREACH_SAFE(head, mq_elt, tmp)
+    DL_FOREACH_SAFE(*head, mq_elt, tmp)
     {
 		topic = (struct stellar_mq_topic_schema *)utarray_eltptr(mq_schema_array,
                                                                     (unsigned int)(mq_elt->topic_id));
@@ -627,10 +628,10 @@ void session_mq_free(struct session *sess, struct stellar_message *head, UT_arra
         {
             topic->sess_msg_free_cb(sess, mq_elt->msg_data, topic->free_cb_arg);
         }
-        DL_DELETE(head, mq_elt);
+        DL_DELETE(*head, mq_elt);
         FREE(mq_elt);
     }
-	FREE(head);
+	FREE(*head);
 }
 
 inline int stellar_session_mq_get_topic_id(struct stellar *st, const char *topic_name)
@@ -720,7 +721,8 @@ int stellar_session_mq_subscribe(struct stellar *st, int topic_id, on_session_ms
 {
 	if(plugin_id >= PACKET_PULGIN_ID_BASE)return -1;// ignore packet plugin
 	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
-	if(plug_mgr->session_mq_schema_array==NULL)return -1;
+	if(plug_mgr == NULL || plug_mgr->session_mq_schema_array==NULL || plug_mgr->registered_session_plugin_array == NULL)return -1;
+
     unsigned int len = utarray_len(plug_mgr->session_mq_schema_array);
     if (len <= (unsigned int)topic_id)return -1;
 
@@ -739,8 +741,18 @@ int stellar_session_mq_subscribe(struct stellar *st, int topic_id, on_session_ms
 	struct stellar_mq_subscriber_info *p=NULL;
   	while( (p=(struct stellar_mq_subscriber_info *)utarray_next(session_plugin_schema->registed_session_mq_subscriber_info,p))) 
 	{
-		if(p->topic_id==topic_id)
-		return 0;
+			struct stellar_mq_subscriber *tmp_subscriber=topic->subscribers;
+			int cnt=0;
+			while(tmp_subscriber)
+			{
+				if(cnt==p->subscriber_idx)
+				{
+					tmp_subscriber->sess_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);
@@ -791,7 +803,15 @@ static void plugin_manager_session_message_dispatch(struct session *sess)
                                 if (session_plugin_schema->on_ctx_new)
                                 {
                                     plugin_ctx_rt->plugin_ctx = session_plugin_schema->on_ctx_new(sess, session_plugin_schema->plugin_env);
-                                    plugin_ctx_rt->state = ACTIVE;
+									if(plugin_ctx_rt->state == EXIT && session_plugin_schema->on_ctx_free)
+									{
+										session_plugin_schema->on_ctx_free(sess, plugin_ctx_rt->plugin_ctx, session_plugin_schema->plugin_env);
+										plugin_ctx_rt->plugin_ctx=NULL;
+									}
+									else
+									{
+                                    	plugin_ctx_rt->state = ACTIVE;
+									}
                                 }
                             }
                             if (sub_elt->sess_msg_cb && bitmap_get(plug_mgr_rt->session_mq_status, mq_elt->topic_id, cur_sub_idx) != 0)// ctx_new maybe call detach, so need check again
@@ -860,15 +880,15 @@ struct plugin_manager_runtime *plugin_manager_session_runtime_new(struct plugin_
 void plugin_manager_session_runtime_free(struct plugin_manager_runtime *rt)
 {
 	if(rt==NULL)return;
+	assert(rt->pending_mq==NULL);
 	if(rt->pending_mq != NULL)
 	{
-		session_mq_free(rt->sess, rt->pending_mq, rt->plug_mgr->session_mq_schema_array);
-		rt->pending_mq=NULL;
+		session_mq_free(rt->sess, &rt->pending_mq, rt->plug_mgr->session_mq_schema_array);
 	}
+	assert(rt->delivered_mq==NULL);
 	if(rt->delivered_mq != NULL)
 	{
-		session_mq_free(rt->sess, rt->delivered_mq, rt->plug_mgr->session_mq_schema_array);
-		rt->delivered_mq=NULL;
+		session_mq_free(rt->sess, &rt->delivered_mq, rt->plug_mgr->session_mq_schema_array);
 	}	
 	if(rt->session_mq_status != NULL)
 	{
@@ -1032,7 +1052,7 @@ void plugin_manager_on_session_ingress(struct session *sess, struct packet *pkt)
 	//TODO: check TCP topic active subscirber num, if 0, return APP_STATE_DROPME, to reduce tcp reassemble overhead
 	session_mq_publish_message(sess, topic_id ,(void *)pkt);
 	plugin_manager_session_message_dispatch(sess);
-	plug_mgr_rt->enable_session_mq=0;
+	//plug_mgr_rt->enable_session_mq=0;
     return;
 }
 
@@ -1040,12 +1060,13 @@ void plugin_manager_on_session_egress(struct session *sess, struct packet *pkt)
 {
 	struct plugin_manager_runtime *plug_mgr_rt = session_plugin_manager_runtime_get(sess);
 	if(plug_mgr_rt==NULL)return;
-	plug_mgr_rt->enable_session_mq=1;
+	//plug_mgr_rt->enable_session_mq=1;
 	session_mq_publish_message(sess, plug_mgr_rt->plug_mgr->egress_topic_id ,(void *)pkt);
 	plugin_manager_session_message_dispatch(sess);
-	session_mq_free(plug_mgr_rt->sess,plug_mgr_rt->delivered_mq, plug_mgr_rt->plug_mgr->session_mq_schema_array);
-	plug_mgr_rt->delivered_mq=NULL;
 	plug_mgr_rt->enable_session_mq=0;
+	session_mq_free(plug_mgr_rt->sess,&plug_mgr_rt->delivered_mq, plug_mgr_rt->plug_mgr->session_mq_schema_array);
+	assert(plug_mgr_rt->pending_mq==NULL);
+	assert(plug_mgr_rt->delivered_mq==NULL);
     return;
 }
 
@@ -1066,12 +1087,14 @@ void stellar_session_plugin_dettach_current_session(struct session *sess)
 		}
 	}	
 
-	if(session_plugin_schema->on_ctx_free)
+	//dettach in ctx INIT, do not call on_ctx_free immidiately
+	if(plug_mgr_rt->plugin_ctx_array[plug_mgr_rt->current_session_plugin_id].state != INIT && session_plugin_schema->on_ctx_free)
 	{
-		session_plugin_schema->on_ctx_free(sess, (plug_mgr_rt->plugin_ctx_array+plug_mgr_rt->current_session_plugin_id)->plugin_ctx, session_plugin_schema->plugin_env);
+		session_plugin_schema->on_ctx_free(sess, plug_mgr_rt->plugin_ctx_array[plug_mgr_rt->current_session_plugin_id].plugin_ctx, session_plugin_schema->plugin_env);
+		plug_mgr_rt->plugin_ctx_array[plug_mgr_rt->current_session_plugin_id].plugin_ctx=NULL;
+		plug_mgr_rt->plugin_ctx_array[plug_mgr_rt->current_session_plugin_id].state=EXIT;
 	}
-	(plug_mgr_rt->plugin_ctx_array+plug_mgr_rt->current_session_plugin_id)->plugin_ctx=NULL;
-	(plug_mgr_rt->plugin_ctx_array+plug_mgr_rt->current_session_plugin_id)->state=EXIT;
+
 	return;
 }
 
diff --git a/src/plugin_manager/plugin_manager_interna.h b/src/plugin_manager/plugin_manager_interna.h
index 0c9e055..24b8d42 100644
--- a/src/plugin_manager/plugin_manager_interna.h
+++ b/src/plugin_manager/plugin_manager_interna.h
@@ -164,6 +164,7 @@ struct registered_session_plugin_schema
 	UT_array *registed_session_mq_subscriber_info;
 }__attribute__((aligned(sizeof(void*))));
 
+#define SESSION_PULGIN_ID_BASE 0x00000
 #define PACKET_PULGIN_ID_BASE 0x10000
 #define POLLING_PULGIN_ID_BASE 0x20000
 
diff --git a/src/stellar_on_sapp/stellar_on_sapp_api.c b/src/stellar_on_sapp/stellar_on_sapp_api.c
index f9e5637..9f4f0ae 100644
--- a/src/stellar_on_sapp/stellar_on_sapp_api.c
+++ b/src/stellar_on_sapp/stellar_on_sapp_api.c
@@ -22,6 +22,7 @@ struct packet
 	enum packet_type type;
 	unsigned char ip_proto;
 	const void *raw_pkt;
+	struct session *sess;
 	struct stellar *st;
 }__attribute__((aligned(sizeof(void*)))) ;
 
@@ -110,6 +111,7 @@ struct session *session_new_on_sapp(struct stellar *st, struct streaminfo *strea
 	sess->session_direction=-1;
     memset(&sess->cur_pkt, 0, sizeof(struct packet));
 	sess->cur_pkt.st=st;
+	sess->cur_pkt.sess=sess;
 	sess->plug_mgr_rt=plugin_manager_session_runtime_new(st->plug_mgr, sess);
     return sess;
 }
@@ -170,6 +172,7 @@ void packet_update_on_sapp(struct stellar *st, struct streaminfo *pstream, void
 	pkt.type=type;
 	pkt.raw_pkt=get_current_rawpkt_from_streaminfo(pstream);
 	pkt.st=st;
+	pkt.sess=NULL;
 	switch (type)
 	{
 	case IPv4:
@@ -187,7 +190,7 @@ void packet_update_on_sapp(struct stellar *st, struct streaminfo *pstream, void
 	default:
 		return;
 	}
-	//TODO: exclude non innermost packet
+	//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)
 	{
@@ -226,8 +229,9 @@ inline enum packet_type packet_get_type(const struct packet *pkt)
 
 int packet_get_direction(const struct packet *pkt)
 {
-	struct session *sess = container_of(pkt, struct session, cur_pkt);
-	assert(sess);
+	struct session *sess = pkt->sess;
+	if(sess==NULL)return PACKET_DIRECTION_UNKNOWN;
+
 	const struct streaminfo *pstream=sess->pstream;
 	if(pstream->curdir==DIR_C2S)
 	{
diff --git a/src/stellar_on_sapp/stellar_on_sapp_loader.c b/src/stellar_on_sapp/stellar_on_sapp_loader.c
index 111c44d..0bbb83a 100644
--- a/src/stellar_on_sapp/stellar_on_sapp_loader.c
+++ b/src/stellar_on_sapp/stellar_on_sapp_loader.c
@@ -74,12 +74,8 @@ char stellar_on_sapp_defer_entry(struct streaminfo *pstream,void **pme, int thre
 	if(sess)
 	{
 		session_defer_on_sapp(sess);
-		return APP_STATE_GIVEME;
-	}
-	else
-	{
-		return APP_STATE_DROPME;
 	}
+	return APP_STATE_GIVEME;
 }
 
 void  STELLAR_DEFER_LOADER_EXIT(void)
diff --git a/test/plugin_manager/plugin_manager_gtest_main.cpp b/test/plugin_manager/plugin_manager_gtest_main.cpp
index 8494e7b..edd8831 100644
--- a/test/plugin_manager/plugin_manager_gtest_main.cpp
+++ b/test/plugin_manager/plugin_manager_gtest_main.cpp
@@ -1,7 +1,8 @@
 #include <gtest/gtest.h>
+#include <limits.h>
 
-#include "plugin_manager_gtest_mock.h"
 #include "stellar/utils.h"
+#include "plugin_manager_gtest_mock.h"
 
 void whitebox_test_plugin_manager_intrisic_metadata(struct stellar *st, struct plugin_manager_schema *plug_mgr)
 {
@@ -56,6 +57,13 @@ void whitebox_test_plugin_manager_intrisic_metadata(struct stellar *st, struct p
 		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_array.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);
+	EXPECT_GE(stellar_session_mq_get_topic_id(st, TOPIC_UDP), 0);
+	EXPECT_GE(stellar_session_mq_get_topic_id(st, TOPIC_EGRESS), 0);
+	EXPECT_GE(stellar_session_mq_get_topic_id(st, TOPIC_CONTROL_PACKET), 0);	
 }
 
 
@@ -71,16 +79,13 @@ TEST(plugin_manager_init, init_without_toml) {
 	plugin_manager_exit(plug_mgr);
 }
 
-static void test_mock_exdata_free(struct packet *pkt, int idx, void *ex_ptr, void *arg)
-{
-	return;
-}
+/******************************************
+ * TEST PLUGIN MANAGER PACKET PLUGIN INIT *
+ ******************************************/
 
-static void test_mock_overwrite_exdata_free(struct packet *pkt, int idx, void *ex_ptr, void *arg)
-{
-	return;
-}
+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};
@@ -88,33 +93,30 @@ TEST(plugin_manager_init, packet_exdata_new_index_overwrite) {
 	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_exdata_free, &st);
+	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_exdata_free, plug_mgr);
+	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);
 
-	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_exdata_free);
-    EXPECT_EQ(exdata_schema->free_arg, &st);
-    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);
+    {
+        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);
+    plugin_manager_exit(plug_mgr);
 }
 
-void test_mock_packet_msg_free(struct packet *pkt, void *msg, void *msg_free_arg)
-{
-	return;
-}
-void test_mock_overwrite_packet_msg_free(struct packet *pkt, void *msg, void *msg_free_arg)
-{
-	return;
-}
+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};
@@ -123,58 +125,63 @@ TEST(plugin_manager_init, packet_mq_topic_create_and_update) {
 
 	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 = (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
+    EXPECT_EQ(stellar_packet_mq_get_topic_id(&st, topic_name), -1); // illegal topic_name
 
-	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);
+    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);
+    }
 
-	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(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(utarray_len(plug_mgr->packet_mq_schema_array), 1);
+    EXPECT_EQ(stellar_packet_mq_update_topic(&st, topic_id, test_mock_overwrite_packet_msg_free, plug_mgr), 0);
 
-	EXPECT_EQ(stellar_packet_mq_destroy_topic(&st, 10), -1);// illgeal topic_id
+    {
+        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, topic_id), 1);
-	EXPECT_EQ(stellar_packet_mq_destroy_topic(&st, topic_id), 0);//duplicate destroy, return -1;
+    EXPECT_EQ(stellar_packet_mq_destroy_topic(&st, 10), -1); // illgeal topic_id
 
-	
-	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);
+    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;
 
-	plugin_manager_exit(plug_mgr);
+    {
+        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)
-{
-	return;
-}
+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)
-{
-	return;
-}
+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) {
 
@@ -195,29 +202,32 @@ TEST(plugin_manager_init, packet_mq_subscribe) {
 	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 = (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_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->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 ON PACKET *
- ***********************************/
+/*******************************************
+ * 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 basic_plugin_env
+struct packet_plugin_env
 {
 	struct plugin_manager_schema *plug_mgr;
 	int basic_on_packet_called;
@@ -236,7 +246,7 @@ struct basic_plugin_env
 
 static void test_basic_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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);
@@ -253,16 +263,19 @@ TEST(plugin_manager, basic_packet_plugin) {
 	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
 
 	unsigned char ip_proto=6;
-	struct basic_plugin_env env;
-	memset(&env, 0, sizeof(struct basic_plugin_env));
+	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);
 
-	int packet_plugin_num=utarray_len(plug_mgr->registered_packet_plugin_array);
-	EXPECT_EQ(packet_plugin_num, 1);
+    {
+        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};
+    struct packet pkt={&st, IPv4, ip_proto};
 	plugin_manager_on_packet_ingress(plug_mgr, &pkt);
 	plugin_manager_on_packet_egress(plug_mgr, &pkt);
 	
@@ -273,7 +286,7 @@ TEST(plugin_manager, basic_packet_plugin) {
 
 static void test_proto_filter_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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);
@@ -289,8 +302,8 @@ TEST(plugin_manager, packet_plugin_proto_filter) {
 	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
 	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
 
-	struct basic_plugin_env env;
-	memset(&env, 0, sizeof(struct basic_plugin_env));
+	struct packet_plugin_env env;
+	memset(&env, 0, sizeof(struct packet_plugin_env));
 	env.plug_mgr=plug_mgr;
 
 	int proto_filter_plugin_num=(int)(sizeof(env.proto_filter_plugin_id) / sizeof(env.proto_filter_plugin_id[0]));
@@ -301,8 +314,11 @@ TEST(plugin_manager, packet_plugin_proto_filter) {
 
 
     }
-	int packet_plugin_num = utarray_len(plug_mgr->registered_packet_plugin_array);
-    EXPECT_EQ(packet_plugin_num, proto_filter_plugin_num);
+
+	{
+        SCOPED_TRACE("White-box test, check stellar internal schema");
+        EXPECT_EQ(utarray_len(plug_mgr->registered_packet_plugin_array), proto_filter_plugin_num);
+    }
 
     struct packet pkt={&st, IPv4, 0};
 
@@ -324,11 +340,9 @@ TEST(plugin_manager, packet_plugin_proto_filter) {
 	}
 }
 
-
-
 static void test_exdata_set_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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);
@@ -347,7 +361,7 @@ static void test_exdata_set_on_packet(struct packet *pkt, unsigned char ip_proto
 
 static void test_exdata_get_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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);
@@ -365,7 +379,7 @@ static void test_exdata_get_on_packet(struct packet *pkt, unsigned char ip_proto
 
 static void test_packet_exdata_free(struct packet *pkt, int idx, void *ex_ptr, void *arg)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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);
@@ -373,7 +387,6 @@ static void test_packet_exdata_free(struct packet *pkt, int idx, void *ex_ptr, v
 	return;
 }
 
-#include <limits.h>
 
 TEST(plugin_manager, packet_exdata) {
 
@@ -382,8 +395,8 @@ TEST(plugin_manager, packet_exdata) {
 	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
 
 	unsigned char ip_proto=6;
-	struct basic_plugin_env env;
-	memset(&env, 0, sizeof(struct basic_plugin_env));
+	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];
@@ -392,27 +405,35 @@ TEST(plugin_manager, packet_exdata) {
 	{
 		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);
-		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");
+            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]);
+        }
+    }
 
-	int exdata_num=utarray_len(plug_mgr->packet_exdata_schema_array);
-	EXPECT_EQ(exdata_num, exdata_idx_len);
+    {
+        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);
+    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);
 
-	int packet_plugin_num=utarray_len(plug_mgr->registered_packet_plugin_array);
-	EXPECT_EQ(packet_plugin_num, 2);//Fix plugin number
+    {
+        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};
+    struct packet pkt={&st, IPv4, ip_proto};
 
 	int N_packet=10;
 
@@ -434,7 +455,7 @@ TEST(plugin_manager, packet_exdata) {
 
 static void test_packet_msg_free_cb_func(struct packet *pkt, void *msg, void *msg_free_arg)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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;
@@ -442,7 +463,7 @@ static void test_packet_msg_free_cb_func(struct packet *pkt, void *msg, void *ms
 
 static void test_mq_on_packet_msg(struct packet *pkt, int topic_id, const void *msg, void *plugin_env)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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;
@@ -451,7 +472,7 @@ static void test_mq_on_packet_msg(struct packet *pkt, int topic_id, const void *
 
 static void test_mq_pub_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
 {
-	struct basic_plugin_env *env = (struct basic_plugin_env *)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);
@@ -471,8 +492,8 @@ TEST(plugin_manager, packet_mq) {
 	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
 
 	unsigned char ip_proto=6;
-	struct basic_plugin_env env;
-	memset(&env, 0, sizeof(struct basic_plugin_env));
+	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];
 
@@ -483,15 +504,21 @@ TEST(plugin_manager, packet_mq) {
 		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);
-		struct stellar_mq_topic_schema *topic = (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->packet_mq_schema_array, env.packet_topic_id[i]);
+        {
+            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]);
+        }
+    }
 
-		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 topic_num=utarray_len(plug_mgr->packet_mq_schema_array);
-	EXPECT_EQ(topic_num, 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);
@@ -508,8 +535,10 @@ TEST(plugin_manager, packet_mq) {
 		}
     }
 
-    int packet_plugin_num=utarray_len(plug_mgr->registered_packet_plugin_array);
-	EXPECT_EQ(packet_plugin_num, topic_sub_num+1);
+	{
+		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};
 
@@ -521,20 +550,521 @@ TEST(plugin_manager, packet_mq) {
     }
 
     plugin_manager_exit(plug_mgr);
-	EXPECT_EQ(N_packet*topic_num, env.msg_pub_cnt);
+	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);
 }
 
-/***********************************
- * TEST PLUGIN MANAGER ON SESSION *
- ***********************************/
+/**********************************************
+ * TEST PLUGIN MANAGER ON SESSION PLUGIN INIT *
+ **********************************************/
 
+static void test_mock_session_exdata_free(struct session *sess, int idx, void *ex_ptr, void *arg){}
+static void test_mock_overwrite_session_exdata_free(struct session *sess, int idx, void *ex_ptr, void *arg){}
 
-/***********************************
- * TEST PLUGIN MANAGER ON POLLING *
- ***********************************/
+TEST(plugin_manager_init, session_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="SESSION_EXDATA";
+	int exdata_idx=stellar_session_exdata_new_index(&st,exdata_name, test_mock_session_exdata_free, &st);
+	EXPECT_GE(exdata_idx, 0);
+	int overwrite_idx=stellar_session_exdata_new_index(&st,exdata_name, test_mock_overwrite_session_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->session_exdata_schema_array, (unsigned int)exdata_idx);
+        EXPECT_EQ(exdata_schema->free_func, (void *)test_mock_overwrite_session_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->session_exdata_schema_array);
+        EXPECT_EQ(exdata_num, 1);
+    }
+    plugin_manager_exit(plug_mgr);
+}
+
+void test_mock_session_msg_free(struct session *sess, void *msg, void *msg_free_arg){}
+void test_mock_overwrite_session_msg_free(struct session *sess, void *msg, void *msg_free_arg){}
+
+TEST(plugin_manager_init, session_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="SESSION_TOPIC";
+
+	EXPECT_EQ(stellar_session_mq_get_topic_id(&st, topic_name), -1);// illegal topic_name
+
+	int topic_id=stellar_session_mq_create_topic(&st, topic_name, test_mock_session_msg_free, &st);
+	EXPECT_GE(topic_id, 0);
+	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->session_mq_schema_array, (unsigned int)topic_id);
+        EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_session_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_session_mq_get_topic_id(&st, topic_name), topic_id);
+	EXPECT_EQ(stellar_session_mq_create_topic(&st, topic_name, test_mock_overwrite_session_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->session_mq_schema_array, (unsigned int)topic_id);
+        EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_session_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_session_mq_update_topic(&st, topic_id, test_mock_overwrite_session_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->session_mq_schema_array, (unsigned int)topic_id);
+        EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_overwrite_session_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->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);
+	EXPECT_EQ(stellar_session_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->session_mq_schema_array), 6);//destory won't delete the topic schema
+	}
+	EXPECT_EQ(plug_mgr->session_mq_topic_num, 5);//intrinsic topic number
+
+	plugin_manager_exit(plug_mgr);
+}
+
+void test_mock_on_session_msg(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env){}
+void test_mock_overwrite_on_session_msg(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env){}
+
+
+TEST(plugin_manager_init, session_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="SESSION_TOPIC";
+
+	int topic_id=stellar_session_mq_create_topic(&st, topic_name, test_mock_session_msg_free, &st);
+	EXPECT_GE(topic_id, 0);
+
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, topic_id, test_mock_on_session_msg, 10),-1);//illgeal plugin_id
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, 10, test_mock_on_session_msg, 10),-1);//illgeal topic_id & plugin_id
+
+	int plugin_id=stellar_session_plugin_register(&st, NULL, NULL, &st);
+	EXPECT_GE(plugin_id, 0);
+
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, topic_id, test_mock_on_session_msg, plugin_id),0);
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, topic_id, test_mock_overwrite_on_session_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->session_mq_schema_array,(unsigned int)topic_id);
+		EXPECT_EQ(topic_schema->free_cb, (void *)test_mock_session_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_session_msg);
+
+	plugin_manager_exit(plug_mgr);
+}
+/**********************************************
+ * TEST PLUGIN MANAGER ON SESSION PLUGIN RUNTIME *
+ **********************************************/
+
+/*******************************************
+ * TEST PLUGIN MANAGER PACKET PLUGIN RUNTIME*
+ *******************************************/
+
+struct session_plugin_env
+{
+	struct plugin_manager_schema *plug_mgr;
+	int N_session;
+	int N_per_session_pkt_cnt;
+	int intrinsc_tcp_topic_id;
+	int intrinsc_egress_topic_id;
+	int basic_exdata_idx;
+	int basic_exdata_free_called;
+	int basic_on_session_ingress_called;
+	int basic_on_session_egress_called;
+	int basic_ctx_new_called;
+	int basic_ctx_free_called;
+	int test_mq_pub_plugin_id;
+	int test_mq_sub_plugin_id;
+	int test_mq_pub_called;
+	int test_mq_sub_called;
+	int test_mq_free_called;
+	int test_mq_topic_id;
+};
+
+struct test_basic_ctx
+{
+	int called;
+};
+
+static void *test_basic_session_ctx_new(struct session *sess, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	env->basic_ctx_new_called+=1;
+	struct test_basic_ctx *ctx=CALLOC(struct test_basic_ctx, 1);
+	return ctx;
+}
+
+static void test_basic_session_ctx_free(struct session *sess, void *session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	env->basic_ctx_free_called+=1;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)session_ctx;
+	EXPECT_EQ(ctx->called, env->N_per_session_pkt_cnt*2);//ingress + egress
+	FREE(ctx);
+	return;
+}
+
+static void test_basic_on_session_ingress(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)per_session_ctx;
+	EXPECT_TRUE(env!=NULL);
+	EXPECT_TRUE(ctx!=NULL);
+	EXPECT_EQ(sess->plug_mgr_rt->plug_mgr, env->plug_mgr);
+	EXPECT_EQ(session_exdata_set(sess, 2, sess), -1);// illegal set
+	EXPECT_EQ(session_exdata_get(sess, 2), nullptr);// illegal get
+	EXPECT_EQ(session_exdata_set(sess, env->basic_exdata_idx, env), 0);
+	env->basic_on_session_ingress_called+=1;
+	ctx->called+=1;
+	return;
+}
+
+static void test_basic_on_session_egress(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)per_session_ctx;
+	EXPECT_TRUE(env!=NULL);
+	EXPECT_TRUE(ctx!=NULL);
+	EXPECT_EQ(sess->plug_mgr_rt->plug_mgr, env->plug_mgr);
+	EXPECT_EQ(session_exdata_set(sess, 2, sess), -1);// illegal set
+	EXPECT_EQ(session_exdata_get(sess, 2), nullptr);// illegal get
+	EXPECT_EQ(session_exdata_get(sess, env->basic_exdata_idx), env);
+	env->basic_on_session_egress_called+=1;
+	ctx->called+=1;
+	return;
+}
+
+static void test_basic_session_exdata_free(struct session *sess, int idx, void *ex_ptr, void *arg)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)arg;
+	EXPECT_EQ(env->basic_exdata_idx, idx);
+	EXPECT_EQ(ex_ptr, env);
+	env->basic_exdata_free_called+=1;
+}
+
+TEST(plugin_manager, basic_session_plugin) {
+
+	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 session_plugin_env env;
+	memset(&env, 0, sizeof(struct session_plugin_env));
+	env.plug_mgr=plug_mgr;
+	env.N_per_session_pkt_cnt=10;
+	env.N_session=10;
+
+	int plugin_id=stellar_session_plugin_register(&st, test_basic_session_ctx_new, test_basic_session_ctx_free, &env);
+	EXPECT_GE(plugin_id, 0);
+
+	env.intrinsc_tcp_topic_id=stellar_session_mq_get_topic_id(&st, TOPIC_TCP);
+	EXPECT_GE(env.intrinsc_tcp_topic_id, 0);
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, env.intrinsc_tcp_topic_id, test_basic_on_session_ingress, plugin_id), 0);
+
+	env.intrinsc_egress_topic_id=stellar_session_mq_get_topic_id(&st, TOPIC_EGRESS);
+	EXPECT_GE(env.intrinsc_egress_topic_id, 0);
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, env.intrinsc_egress_topic_id, test_basic_on_session_ingress, plugin_id), 0);// Intentional error
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, env.intrinsc_egress_topic_id, test_basic_on_session_egress, plugin_id), 0);
+
+	env.basic_exdata_idx=stellar_session_exdata_new_index(&st, "SESSION_EXDATA", test_basic_session_exdata_free,&env);
+	EXPECT_GE(env.basic_exdata_idx, 0);
+
+	struct packet pkt={&st, TCP, ip_proto};
+
+
+	struct session sess[env.N_session];
+
+	for(int i=0; i < env.N_session; i++)
+		sess[i].plug_mgr_rt=plugin_manager_session_runtime_new(plug_mgr, &sess[i]);
+
+    for (int j = 0; j < env.N_per_session_pkt_cnt; j++)
+    {
+        plugin_manager_on_packet_ingress(plug_mgr, &pkt);
+        plugin_manager_on_packet_egress(plug_mgr, &pkt);
+
+        for (int i = 0; i < env.N_session; i++)
+        {
+            plugin_manager_on_session_ingress(&sess[i], &pkt);
+            plugin_manager_on_session_egress(&sess[i], &pkt);
+        }
+    }
+
+    for(int i=0; i < env.N_session; i++)
+		plugin_manager_session_runtime_free(sess[i].plug_mgr_rt);
+
+	plugin_manager_exit(plug_mgr);
+
+	EXPECT_TRUE(env.basic_on_session_ingress_called == env.basic_on_session_egress_called && env.basic_on_session_ingress_called == env.N_session*env.N_per_session_pkt_cnt);
+	EXPECT_TRUE(env.basic_ctx_new_called == env.basic_ctx_free_called && env.basic_ctx_new_called == env.N_session);
+	EXPECT_EQ(env.basic_exdata_free_called, env.N_session);
+
+}
+
+struct test_session_mq_ctx
+{
+	int called;
+};
+
+static void *test_mq_pub_session_ctx_new(struct session *sess, void *plugin_env)
+{
+	struct test_basic_ctx *ctx=CALLOC(struct test_basic_ctx, 1);
+	return ctx;
+}
+
+static void test_mq_pub_session_ctx_free(struct session *sess, void *session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)session_ctx;
+	EXPECT_EQ(ctx->called, env->N_per_session_pkt_cnt);
+	FREE(ctx);
+	return;
+}
+
+static void *test_mq_sub_session_ctx_new(struct session *sess, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	struct test_basic_ctx *ctx=CALLOC(struct test_basic_ctx, 1);
+	EXPECT_EQ(session_mq_ignore_message(sess, env->intrinsc_tcp_topic_id, env->test_mq_sub_plugin_id), 0);
+	return ctx;
+}
+
+static void test_mq_sub_session_ctx_free(struct session *sess, void *session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)session_ctx;
+	EXPECT_EQ(ctx->called, env->N_per_session_pkt_cnt);
+	FREE(ctx);
+	return;
+}
+
+static void test_mq_pub_on_session(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)per_session_ctx;
+	EXPECT_TRUE(env!=NULL);
+	EXPECT_TRUE(ctx!=NULL);
+	EXPECT_EQ(sess->plug_mgr_rt->plug_mgr, env->plug_mgr);
+	env->test_mq_pub_called+=1;
+	ctx->called+=1;
+	int *pub_msg=(int *)CALLOC(int, 1);
+	*pub_msg=env->test_mq_pub_called;
+	EXPECT_EQ(session_mq_publish_message(sess, env->test_mq_topic_id, pub_msg),0);
+	return;
+}
+
+static void test_mq_on_sub_msg(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)per_session_ctx;
+	EXPECT_TRUE(env!=NULL);
+	EXPECT_TRUE(ctx!=NULL);
+	EXPECT_EQ(sess->plug_mgr_rt->plug_mgr, env->plug_mgr);
+	EXPECT_EQ(*(int *)msg, env->test_mq_pub_called);
+	env->test_mq_sub_called+=1;
+	ctx->called+=1;
+	return;
+}
+
+static void test_session_msg_free(struct session *sess, void *msg, void *msg_free_arg)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)msg_free_arg;
+	EXPECT_EQ(sess->plug_mgr_rt->plug_mgr, env->plug_mgr);
+	EXPECT_EQ(env->test_mq_pub_called, *(int *)msg);
+	env->test_mq_free_called+=1;
+	FREE(msg);
+	return;
+}
+
+TEST(plugin_manager, basic_session_plugin_mq) {
+
+	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 session_plugin_env env;
+	memset(&env, 0, sizeof(struct session_plugin_env));
+	env.plug_mgr=plug_mgr;
+	env.N_per_session_pkt_cnt=10;
+	env.N_session=10;
+
+	env.test_mq_pub_plugin_id=stellar_session_plugin_register(&st, test_mq_pub_session_ctx_new, test_mq_pub_session_ctx_free, &env);
+	EXPECT_GE(env.test_mq_pub_plugin_id, 0);
+
+	env.intrinsc_tcp_topic_id=stellar_session_mq_get_topic_id(&st, TOPIC_TCP);
+	EXPECT_GE(env.intrinsc_tcp_topic_id, 0);
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, env.intrinsc_tcp_topic_id, test_mq_pub_on_session, env.test_mq_pub_plugin_id), 0);
+
+	env.test_mq_topic_id=stellar_session_mq_create_topic(&st, "SESSION_MQ_TOPIC", test_session_msg_free, &env);
+	EXPECT_GE(env.test_mq_topic_id, 0);
+
+	env.test_mq_sub_plugin_id=stellar_session_plugin_register(&st, test_mq_sub_session_ctx_new, test_mq_sub_session_ctx_free, &env);
+	EXPECT_GE(env.test_mq_sub_plugin_id, 0);
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, env.test_mq_topic_id, test_mq_on_sub_msg, env.test_mq_sub_plugin_id), 0);
+
+	struct packet pkt={&st, TCP, ip_proto};
+
+
+	struct session sess[env.N_session];
+
+	for(int i=0; i < env.N_session; i++)
+		sess[i].plug_mgr_rt=plugin_manager_session_runtime_new(plug_mgr, &sess[i]);
+
+    for (int j = 0; j < env.N_per_session_pkt_cnt; j++)
+    {
+        plugin_manager_on_packet_ingress(plug_mgr, &pkt);
+        plugin_manager_on_packet_egress(plug_mgr, &pkt);
+
+        for (int i = 0; i < env.N_session; i++)
+        {
+            plugin_manager_on_session_ingress(&sess[i], &pkt);
+            plugin_manager_on_session_egress(&sess[i], &pkt);
+        }
+    }
+
+    for(int i=0; i < env.N_session; i++)
+		plugin_manager_session_runtime_free(sess[i].plug_mgr_rt);
+
+	plugin_manager_exit(plug_mgr);
+
+	EXPECT_TRUE(env.test_mq_free_called == env.test_mq_pub_called && env.test_mq_sub_called == env.N_session*env.N_per_session_pkt_cnt);
+
+}
+
+
+
+static void *test_dettach_session_ctx_new(struct session *sess, void *plugin_env)
+{
+	struct test_basic_ctx *ctx=CALLOC(struct test_basic_ctx, 1);
+	stellar_session_plugin_dettach_current_session(sess);
+	ctx->called+=1;
+	return ctx;
+}
+
+static void test_dettach_session_ctx_free(struct session *sess, void *session_ctx, void *plugin_env)
+{
+	struct session_plugin_env *env = (struct session_plugin_env *)plugin_env;
+	env->basic_ctx_free_called+=1;
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)session_ctx;
+	EXPECT_EQ(ctx->called, 1);
+	FREE(ctx);
+}
+
+static void test_dettach_on_session(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env)
+{
+	struct test_basic_ctx *ctx=(struct test_basic_ctx *)per_session_ctx;
+	ctx->called+=1;
+}
+
+TEST(plugin_manager, session_plugin_ctx_new_dettach) {
+
+	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 session_plugin_env env;
+	memset(&env, 0, sizeof(struct session_plugin_env));
+	env.plug_mgr=plug_mgr;
+	env.N_per_session_pkt_cnt=10;
+	env.N_session=10;
+
+
+	int plugin_id=stellar_session_plugin_register(&st, test_dettach_session_ctx_new, test_dettach_session_ctx_free, &env);
+	EXPECT_GE(plugin_id,0);
+
+	env.intrinsc_tcp_topic_id=stellar_session_mq_get_topic_id(&st, TOPIC_TCP);
+	EXPECT_GE(env.intrinsc_tcp_topic_id, 0);
+	EXPECT_EQ(stellar_session_mq_subscribe(&st, env.intrinsc_tcp_topic_id, test_dettach_on_session, env.test_mq_pub_plugin_id), 0);
+
+	struct packet pkt={&st, TCP, ip_proto};
+
+
+	struct session sess[env.N_session];
+
+	for(int i=0; i < env.N_session; i++)
+		sess[i].plug_mgr_rt=plugin_manager_session_runtime_new(plug_mgr, &sess[i]);
+
+    for (int j = 0; j < env.N_per_session_pkt_cnt; j++)
+    {
+        plugin_manager_on_packet_ingress(plug_mgr, &pkt);
+        plugin_manager_on_packet_egress(plug_mgr, &pkt);
+
+        for (int i = 0; i < env.N_session; i++)
+        {
+            plugin_manager_on_session_ingress(&sess[i], &pkt);
+            plugin_manager_on_session_egress(&sess[i], &pkt);
+        }
+    }
+
+    for(int i=0; i < env.N_session; i++)
+		plugin_manager_session_runtime_free(sess[i].plug_mgr_rt);
+
+	plugin_manager_exit(plug_mgr);
+
+	EXPECT_EQ(env.basic_ctx_free_called,env.N_session);
+}
+
+// TODO: test case, session_plugin_ctx_free_send_msg
+TEST(plugin_manager, session_plugin_ctx_free_send_msg) {
+
+	struct stellar st={0};
+	struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
+	whitebox_test_plugin_manager_intrisic_metadata(&st, plug_mgr);
+	plugin_manager_exit(plug_mgr);
+}
+
+/**********************************************
+ * TEST PLUGIN MANAGER ON POLLING PLUGIN INIT *
+ **********************************************/
 
+/**********************************************
+ * TEST PLUGIN MANAGER ON POLLING PLUGIN RUNTIME *
+ **********************************************/
 
 int main(int argc, char ** argv)
 {