gfwleak/stellar-stellar
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ feat(plugin_manager): add session plugin with hook

Browse Source
This commit is contained in:
yangwei
2024-09-04 14:44:30 +08:00
parent c494bf831e
commit 1bdfc4506c
6 changed files with 126 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
infra/plugin_manager/plugin_manager.c
View File

@@ -513,6 +513,28 @@ static int stellar_mq_subscribe(struct plugin_manager_schema *plug_mgr, int topi
return 0;
}
static void plugin_manager_runtime_update_plugin_ctx(struct session *sess, struct registered_session_plugin_schema *session_plugin_schema, struct session_plugin_ctx_runtime *plugin_ctx_rt)
{
if(sess==NULL || session_plugin_schema == NULL || plugin_ctx_rt == NULL)return;
if (plugin_ctx_rt->state == INIT)
{
if (session_plugin_schema->on_ctx_new)
{
plugin_ctx_rt->plugin_ctx = session_plugin_schema->on_ctx_new(sess, session_plugin_schema->plugin_env);
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;
}
}
}
}
static void stellar_mq_dispatch_one_session_message(struct session *sess, struct stellar_message *mq_elt)
{
struct plugin_manager_runtime *plug_mgr_rt = (struct plugin_manager_runtime *)session_get_user_data(sess);
@@ -535,24 +557,7 @@ static void stellar_mq_dispatch_one_session_message(struct session *sess, struct
plug_mgr_rt->plug_mgr->registered_session_plugin_array, (unsigned int)sub_elt->plugin_idx);
if (session_plugin_schema)
{
if (plugin_ctx_rt->state == INIT)
{
if (session_plugin_schema->on_ctx_new)
{
plugin_ctx_rt->plugin_ctx =
session_plugin_schema->on_ctx_new(sess, session_plugin_schema->plugin_env);
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;
}
}
}
plugin_manager_runtime_update_plugin_ctx(sess, session_plugin_schema, plugin_ctx_rt);
if (sub_elt->sess_msg_cb &&
bitmap_get(plug_mgr_rt->session_mq_status, cur_sub_idx, mq_elt->header.topic_id) !=
0) // ctx_new maybe call detach, need check again
@@ -973,9 +978,11 @@ int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr)
*********************************************/
UT_icd registered_session_plugin_schema_icd = {sizeof(struct registered_session_plugin_schema), NULL, NULL, NULL};
int stellar_session_plugin_register(struct stellar *st,
int stellar_session_plugin_register_with_hook(struct stellar *st,
session_ctx_new_func session_ctx_new,
session_ctx_free_func session_ctx_free,
session_on_new_func session_on_new,
session_on_free_func session_on_free,
void *plugin_env)
{
struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
@@ -987,11 +994,21 @@ int stellar_session_plugin_register(struct stellar *st,
memset(&session_plugin_schema, 0, sizeof(struct registered_session_plugin_schema));
session_plugin_schema.on_ctx_new = session_ctx_new;
session_plugin_schema.on_ctx_free = session_ctx_free;
session_plugin_schema.on_session_new = session_on_new;
session_plugin_schema.on_session_free = session_on_free;
session_plugin_schema.plugin_env = plugin_env;
utarray_push_back(plug_mgr->registered_session_plugin_array, &session_plugin_schema);
return (utarray_len(plug_mgr->registered_session_plugin_array)-1);// return session plugin_id, equals to session plugin arrary index
}
int stellar_session_plugin_register(struct stellar *st,
session_ctx_new_func session_ctx_new,
session_ctx_free_func session_ctx_free,
void *plugin_env)
{
return stellar_session_plugin_register_with_hook(st, session_ctx_new, session_ctx_free, NULL, NULL, plugin_env);
}
void plugin_manager_on_session_input(struct session *sess, struct packet *pkt)
{
if(sess==NULL)return;
@@ -1064,30 +1081,54 @@ void plugin_manager_on_session_output(struct session *sess, struct packet *pkt)
return;
}
void plugin_manager_on_session_closing(struct session *sess)
void plugin_manager_on_session_new(struct plugin_manager_schema *plug_mgr, struct session *sess)
{
if(sess==NULL)return;
if(plug_mgr->registered_session_plugin_array==NULL)return;
struct plugin_manager_runtime *plug_mgr_rt = plugin_manager_session_runtime_new(plug_mgr, sess);
session_set_user_data(sess, plug_mgr_rt);
struct registered_session_plugin_schema *s = NULL;
struct session_plugin_ctx_runtime *plugin_ctx_rt;
for(unsigned int i = 0; i < utarray_len(plug_mgr_rt->plug_mgr->registered_session_plugin_array); i++)
{
s = (struct registered_session_plugin_schema *)utarray_eltptr(plug_mgr_rt->plug_mgr->registered_session_plugin_array, i);
if (s->on_session_new)
{
plugin_ctx_rt = (plug_mgr_rt->plugin_ctx_array + i);
plugin_manager_runtime_update_plugin_ctx(sess, s, plugin_ctx_rt);
s->on_session_new(sess, plugin_ctx_rt->plugin_ctx, s->plugin_env);
}
}
return;
}
void plugin_manager_on_session_free(struct session *sess)
{
if(sess==NULL)return;
struct plugin_manager_runtime *plug_mgr_rt = (struct plugin_manager_runtime *)session_get_user_data(sess);
if(plug_mgr_rt==NULL)return;
if(plug_mgr_rt->plug_mgr->registered_session_plugin_array==NULL)return;
plug_mgr_rt->pub_session_msg_cnt=0;// reset pub_msg_cnt
switch (session_get_type(sess))
{
case SESSION_TYPE_TCP:
session_mq_publish_message_with_priority(sess, plug_mgr_rt->plug_mgr->tcp_input_topic_id ,NULL, STELLAR_MQ_PRIORITY_HIGH);
session_mq_publish_message_with_priority(sess, plug_mgr_rt->plug_mgr->tcp_stream_topic_id , NULL, STELLAR_MQ_PRIORITY_HIGH);
session_mq_publish_message_with_priority(sess, plug_mgr_rt->plug_mgr->tcp_output_topic_id , NULL, STELLAR_MQ_PRIORITY_HIGH);
break;
case SESSION_TYPE_UDP:
session_mq_publish_message_with_priority(sess, plug_mgr_rt->plug_mgr->udp_input_topic_id ,NULL, STELLAR_MQ_PRIORITY_HIGH);
session_mq_publish_message_with_priority(sess, plug_mgr_rt->plug_mgr->udp_output_topic_id ,NULL, STELLAR_MQ_PRIORITY_HIGH);
break;
default:
break;
}
int tid=stellar_get_current_thread_index();
struct registered_session_plugin_schema *s = NULL;
struct session_plugin_ctx_runtime *plugin_ctx_rt;
for(unsigned int i = 0; i < utarray_len(plug_mgr_rt->plug_mgr->registered_session_plugin_array); i++)
{
s = (struct registered_session_plugin_schema *)utarray_eltptr(plug_mgr_rt->plug_mgr->registered_session_plugin_array, i);
if (s->on_session_free)
{
plugin_ctx_rt = (plug_mgr_rt->plugin_ctx_array + i);
plugin_manager_runtime_update_plugin_ctx(sess, s, plugin_ctx_rt);
s->on_session_free(sess, plugin_ctx_rt->plugin_ctx, s->plugin_env);
}
}
int tid=stellar_get_current_thread_index();
stellar_mq_dispatch(plug_mgr_rt->plug_mgr->per_thread_data[tid].priority_mq, &plug_mgr_rt->plug_mgr->per_thread_data[tid].dealth_letter_queue, sess, NULL);
plug_mgr_rt->pub_session_msg_cnt=-1;//disable session message publish
stellar_mq_free(&plug_mgr_rt->plug_mgr->per_thread_data[tid].dealth_letter_queue, plug_mgr_rt->plug_mgr->stellar_mq_schema_array);
plugin_manager_session_runtime_free(plug_mgr_rt);
return;
}