🦄 refactor(polling manager): merge polling into module manager

3 files changed, 106 insertions, 0 deletions

diff --git a/infra/module_manager/module_manager.c b/infra/module_manager/module_manager.c
index 517fdea..f05256e 100644
--- a/infra/module_manager/module_manager.c
+++ b/infra/module_manager/module_manager.c
@@ -226,3 +226,50 @@ void stellar_module_set_name(struct stellar_module* mod, const char *name)
 	return;
 }
 
+/*******************************************
+ * polling API *
+ *******************************************/
+
+ #define TOPIC_POLLING "polling"
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wcast-function-type"
+static void on_polling_dispatch(int topic_id __unused, 
+									 void *msg __unused,  
+									on_msg_cb_func* on_msg_cb, 
+									 void *on_msg_cb_arg, 
+									 void *dispatch_arg)
+{
+	struct stellar_module_manager *mod_mgr=(struct stellar_module_manager *)dispatch_arg;
+	module_on_polling_func *polling = (module_on_polling_func *)on_msg_cb;
+	polling(mod_mgr, on_msg_cb_arg);
+}
+
+int stellar_module_manager_polling_subscribe(struct stellar_module_manager	*mod_mgr,  module_on_polling_func on_polling, void *polling_arg)
+{
+	if(mod_mgr == NULL)return -1;
+	mod_mgr->topic_polling_id=mq_schema_get_topic_id(stellar_module_manager_get_mq_schema(mod_mgr), TOPIC_POLLING);
+	if(mod_mgr->topic_polling_id<0)
+	{
+		mod_mgr->topic_polling_id=mq_schema_create_topic(mod_mgr->schema.mq_schema, TOPIC_POLLING, on_polling_dispatch, mod_mgr, NULL, NULL);
+	}
+	return mq_schema_subscribe(mod_mgr->schema.mq_schema, mod_mgr->topic_polling_id, (on_msg_cb_func *)on_polling, polling_arg);
+}
+
+#pragma GCC diagnostic pop 
+
+void stellar_module_manager_polling_active(struct stellar_module_manager *mod_mgr)
+{
+	if(mod_mgr == NULL)return;
+	mq_runtime_publish_message(local_mq_rt, mod_mgr->topic_polling_id, NULL);
+}
+
+
+void stellar_polling_dispatch(struct stellar_module_manager *mod_mgr)
+{
+	if(mod_mgr==NULL)return;
+	stellar_module_manager_polling_active(mod_mgr);
+	mq_runtime_dispatch(local_mq_rt);
+	mq_runtime_clean(local_mq_rt);
+	return;
+}
\ No newline at end of file
diff --git a/infra/module_manager/module_manager_interna.h b/infra/module_manager/module_manager_interna.h
index 96b46a7..b58a374 100644
--- a/infra/module_manager/module_manager_interna.h
+++ b/infra/module_manager/module_manager_interna.h
@@ -36,6 +36,7 @@ struct stellar_module_manager
     char *module_spec_toml_path;
     struct module_spec_load *module_specs;
     int load_module_num;
+    int topic_polling_id;
     struct
     {
         int max_thread_num;
@@ -45,6 +46,7 @@ struct stellar_module_manager
     
 }__attribute__((aligned(sizeof(void*))));
 
+void stellar_polling_dispatch(struct stellar_module_manager *mod_mgr);
 
 #ifdef __cplusplus
 }
diff --git a/infra/module_manager/test/gtest_module_manager_main.cpp b/infra/module_manager/test/gtest_module_manager_main.cpp
index 20d5b6d..5a30553 100644
--- a/infra/module_manager/test/gtest_module_manager_main.cpp
+++ b/infra/module_manager/test/gtest_module_manager_main.cpp
@@ -182,6 +182,63 @@ TEST(module_manager, basic_module) {
 	unlink(toml_template);
 }
 
+/***********************************
+ * TEST POLLING MANAGER POLLING API *
+ ***********************************/
+
+struct test_module_polling_env
+{
+	int N_round;
+	int polling_count;
+	int polling_active_count;
+};
+
+ void test_module_on_polling(struct stellar_module_manager* mod_mgr, void *polling_arg)
+ {
+	struct test_module_polling_env *env = (struct test_module_polling_env*)polling_arg;
+	env->polling_count++;
+	if(env->polling_count%2==0)
+	{
+		stellar_module_manager_polling_active(mod_mgr);
+		env->polling_active_count++;
+	}
+ }
+
+
+TEST(polling_manager, basic_polling_module) {
+
+	struct mq_schema *mq_schema=mq_schema_new();
+
+	struct stellar_module_manager *mod_mgr=stellar_module_manager_new(NULL, 10, mq_schema, NULL);
+	EXPECT_TRUE(mod_mgr!=NULL);
+
+
+	EXPECT_EQ(stellar_module_manager_get_max_thread_num(mod_mgr), 10);
+	EXPECT_EQ(stellar_module_manager_get_mq_schema(mod_mgr), mq_schema);
+
+	struct test_module_polling_env env={};
+	env.N_round=10;
+
+	stellar_module_manager_polling_subscribe(mod_mgr, test_module_on_polling, &env);
+
+	struct mq_runtime *mq_rt = mq_runtime_new(mq_schema);
+	stellar_module_manager_register_thread(mod_mgr, 1, mq_rt);
+	
+	EXPECT_EQ((long)stellar_module_manager_get_thread_id(mod_mgr), 1);
+	EXPECT_EQ(stellar_module_manager_get_mq_runtime(mod_mgr), mq_rt);
+
+	for(int i=0; i<env.N_round; i++)
+	{
+		stellar_polling_dispatch(mod_mgr);
+	}
+
+	stellar_module_manager_free(mod_mgr);	
+	
+	EXPECT_EQ(env.polling_count, env.N_round+env.polling_active_count);
+
+
+}
+
 /**********************************************
  * GTEST MAIN *
  **********************************************/