✨ feat(module manager): Define and implement the polling API

6 files changed, 102 insertions, 7 deletions

diff --git a/include/stellar/module_manager.h b/include/stellar/module_manager.h
index 06c51a4..7d99bec 100644
--- a/include/stellar/module_manager.h
+++ b/include/stellar/module_manager.h
@@ -37,6 +37,11 @@ int stellar_module_manager_get_max_thread_num(struct stellar_module_manager* mod
 const char *stellar_module_manager_get_toml_path(struct stellar_module_manager *mod_mgr);
 struct mq_schema *stellar_module_manager_get_mq_schema(struct stellar_module_manager *mod_mgr);
 
+typedef void module_on_polling_func(struct stellar_module_manager* mod_mgr, void *polling_arg);
+//return 0 if success, otherwise return -1.
+int stellar_module_manager_polling_subscribe(struct stellar_module_manager* mod_mgr,  module_on_polling_func on_polling, void *polling_arg);
+void stellar_module_manager_polling_dispatch(struct stellar_module_manager *mod_mgr);
+void stellar_module_manager_polling_active(struct stellar_module_manager *mod_mgr);
 
 #ifdef __cplusplus
 }
diff --git a/include/stellar/stellar.h b/include/stellar/stellar.h
index a2b6f28..e03b791 100644
--- a/include/stellar/stellar.h
+++ b/include/stellar/stellar.h
@@ -18,11 +18,6 @@ typedef void plugin_on_packet_func(struct packet *pkt, void *on_packet_cb_arg);
 //return 0 if success, otherwise return -1.
 int stellar_raw_packet_subscribe(struct stellar *st, plugin_on_packet_func *on_packet_cb, void *on_packet_cb_arg);
 
-//return on_polling state, 0: idle, 1: working
-typedef int plugin_on_polling_func(void *polling_arg);
-//return 0 if success, otherwise return -1.
-int stellar_polling_subscribe(struct stellar *st,  plugin_on_polling_func on_polling, void *polling_arg);
-
 void stellar_emit_datapath_telemetry(struct packet *pkt,  const char * module, const char *str);
 
 // only send user build packet, can't send packet which come from network
diff --git a/infra/module_manager/CMakeLists.txt b/infra/module_manager/CMakeLists.txt
index dddde31..131cf93 100644
--- a/infra/module_manager/CMakeLists.txt
+++ b/infra/module_manager/CMakeLists.txt
@@ -3,6 +3,6 @@ target_include_directories(module_manager PUBLIC ${CMAKE_CURRENT_LIST_DIR})
 target_include_directories(module_manager PUBLIC ${CMAKE_SOURCE_DIR}/include/)
 target_include_directories(module_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra/)
 target_include_directories(module_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/)
-target_link_libraries(module_manager PUBLIC toml ${CMAKE_DL_LIBS})
+target_link_libraries(module_manager PUBLIC toml mq ${CMAKE_DL_LIBS})
 
 add_subdirectory(test)
\ No newline at end of file
diff --git a/infra/module_manager/module_manager.c b/infra/module_manager/module_manager.c
index 3006bee..38e2135 100644
--- a/infra/module_manager/module_manager.c
+++ b/infra/module_manager/module_manager.c
@@ -236,4 +236,43 @@ void stellar_module_set_name(struct stellar_module* mod, const char *name)
 	if(mod==NULL)return;
 	memcpy(mod->name, name, MIN(NAME_MAX, strlen(name)));
 	return;
-}
\ No newline at end of file
+}
+#define TOPIC_POLLING "POLLING"
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wcast-function-type"
+static void stellar_module_manager_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)
+{
+	mod_mgr->schema.polling_topic_id=mq_schema_get_topic_id(mod_mgr->schema.mq_schema, TOPIC_POLLING);
+	if(mod_mgr->schema.polling_topic_id<0)
+	{
+		mod_mgr->schema.polling_topic_id=mq_schema_create_topic(mod_mgr->schema.mq_schema, TOPIC_POLLING, stellar_module_manager_on_polling_dispatch, mod_mgr, NULL, NULL);
+	}
+	return mq_schema_subscribe(mod_mgr->schema.mq_schema, mod_mgr->schema.polling_topic_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)
+{
+	mq_runtime_publish_message(local_mq_rt, mod_mgr->schema.polling_topic_id, NULL);
+}
+
+void stellar_module_manager_polling_dispatch(struct stellar_module_manager *mod_mgr)
+{
+	stellar_module_manager_polling_active(mod_mgr);
+	mq_runtime_dispatch(local_mq_rt);
+	mq_runtime_clean(local_mq_rt);
+	return;
+}
diff --git a/infra/module_manager/module_manager_interna.h b/infra/module_manager/module_manager_interna.h
index 7d28b50..5d571a6 100644
--- a/infra/module_manager/module_manager_interna.h
+++ b/infra/module_manager/module_manager_interna.h
@@ -40,6 +40,7 @@ struct stellar_module_manager
         int load_module_num;
         int max_thread_num;
         struct mq_schema *mq_schema;
+        int polling_topic_id;
     }schema;
     
 }__attribute__((aligned(sizeof(void*))));
diff --git a/infra/module_manager/test/gtest_module_manager_main.cpp b/infra/module_manager/test/gtest_module_manager_main.cpp
index 74aa611..e46037b 100644
--- a/infra/module_manager/test/gtest_module_manager_main.cpp
+++ b/infra/module_manager/test/gtest_module_manager_main.cpp
@@ -200,6 +200,61 @@ TEST(module_manager, basic_module) {
 
 }
 
+/***********************************
+ * TEST MODULE 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(module_manager_polling, 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);
+	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_module_manager_polling_dispatch(mod_mgr);
+	}
+
+	stellar_module_manager_free(mod_mgr);	
+
+	EXPECT_EQ(env.polling_count, env.N_round+env.polling_active_count);
+
+}
+
 /**********************************************
  * GTEST MAIN *
  **********************************************/