✨ feat(plugin manager intrisic mq): send empty msg when session closingFeature-closing-send-sesison-msg

3 files changed, 28 insertions, 3 deletions

diff --git a/src/plugin_manager/plugin_manager.c b/src/plugin_manager/plugin_manager.c
index 2cd5a81..2f20d51 100644
--- a/src/plugin_manager/plugin_manager.c
+++ b/src/plugin_manager/plugin_manager.c
@@ -1066,12 +1066,35 @@ 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;
-	session_mq_publish_message(sess, plug_mgr_rt->plug_mgr->egress_topic_id ,(void *)pkt);
+	session_mq_publish_message(sess, plug_mgr_rt->plug_mgr->egress_topic_id ,pkt);
 	plugin_manager_session_message_dispatch(sess);
 	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);
+	assert(plug_mgr_rt->pending_mq==NULL && plug_mgr_rt->delivered_mq==NULL);
+	return;
+}
+
+void plugin_manager_on_session_closing(struct session *sess)
+{
+	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;
+	switch (session_get_type(sess))
+	{
+		case SESSION_TYPE_TCP:
+			session_mq_publish_message(sess, plug_mgr_rt->plug_mgr->tcp_topic_id ,NULL);
+			session_mq_publish_message(sess, plug_mgr_rt->plug_mgr->tcp_stream_topic_id , NULL);
+			break;
+		case SESSION_TYPE_UDP:
+			session_mq_publish_message(sess, plug_mgr_rt->plug_mgr->udp_topic_id ,NULL);
+			break;
+		default:
+			break;
+	}
+	plugin_manager_session_message_dispatch(sess);
+	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 && plug_mgr_rt->delivered_mq==NULL);
     return;
 }
 
diff --git a/src/plugin_manager/plugin_manager.h b/src/plugin_manager/plugin_manager.h
index 1305f19..6810da8 100644
--- a/src/plugin_manager/plugin_manager.h
+++ b/src/plugin_manager/plugin_manager.h
@@ -17,6 +17,7 @@ int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr);
 //publish and dispatch session msg(msg, pkt) on session_mq
 void plugin_manager_on_session_ingress(struct session *sess,struct packet *pkt);
 void plugin_manager_on_session_egress(struct session *sess,struct packet *pkt);
+void plugin_manager_on_session_closing(struct session *sess);
 
 struct plugin_manager_runtime *plugin_manager_session_runtime_new(struct plugin_manager_schema *plug_mgr, struct session *sess);
 void plugin_manager_session_runtime_free(struct plugin_manager_runtime *plug_mgr_rt);
\ No newline at end of file
diff --git a/src/stellar_on_sapp/stellar_on_sapp_api.c b/src/stellar_on_sapp/stellar_on_sapp_api.c
index a7b590a..9f2794d 100644
--- a/src/stellar_on_sapp/stellar_on_sapp_api.c
+++ b/src/stellar_on_sapp/stellar_on_sapp_api.c
@@ -119,6 +119,7 @@ struct session *session_new_on_sapp(struct stellar *st, struct streaminfo *strea
 void session_free_on_sapp(struct session *sess)
 {
 	sess->state = SESSION_STATE_CLOSING;
+	plugin_manager_on_session_closing(sess);
 	if(sess->plug_mgr_rt)
 	{
 		plugin_manager_session_runtime_free(sess->plug_mgr_rt);