✨ feat(polling in stellar core): worker thread call polling_dispatch

3 files changed, 10 insertions, 3 deletions

diff --git a/infra/polling_manager/polling_manager.c b/infra/polling_manager/polling_manager.c
index 5d9013a..81a585b 100644
--- a/infra/polling_manager/polling_manager.c
+++ b/infra/polling_manager/polling_manager.c
@@ -46,6 +46,7 @@ static void on_polling_dispatch(int topic_id __unused,
 
 int stellar_polling_subscribe(struct stellar_polling_manager* polling_mgr,  module_on_polling_func on_polling, void *polling_arg)
 {
+	if(polling_mgr==NULL || polling_mgr->mod_mgr == NULL)return -1;
 	polling_mgr->polling_topic_id=mq_schema_get_topic_id(stellar_module_manager_get_mq_schema(polling_mgr->mod_mgr), TOPIC_POLLING);
 	if(polling_mgr->polling_topic_id<0)
 	{
@@ -58,11 +59,13 @@ int stellar_polling_subscribe(struct stellar_polling_manager* polling_mgr,  modu
 
 void stellar_polling_active(struct stellar_polling_manager *polling_mgr)
 {
+	if(polling_mgr==NULL || polling_mgr->mod_mgr == NULL)return;
 	mq_runtime_publish_message(stellar_module_manager_get_mq_runtime(polling_mgr->mod_mgr), polling_mgr->polling_topic_id, NULL);
 }
 
 void stellar_polling_dispatch(struct stellar_polling_manager *polling_mgr)
 {
+	if(polling_mgr==NULL || polling_mgr->mod_mgr == NULL)return;
 	stellar_polling_active(polling_mgr);
 	mq_runtime_dispatch(stellar_module_manager_get_mq_runtime(polling_mgr->mod_mgr));
 	mq_runtime_clean(stellar_module_manager_get_mq_runtime(polling_mgr->mod_mgr));
diff --git a/infra/polling_manager/polling_manager_internal.h b/infra/polling_manager/polling_manager_internal.h
index e6c02dd..c0d2f1b 100644
--- a/infra/polling_manager/polling_manager_internal.h
+++ b/infra/polling_manager/polling_manager_internal.h
@@ -15,6 +15,7 @@ struct stellar_polling_manager
 	int polling_topic_id;
 };
 
+//TODO: expose this function to polling_manager.h
 void stellar_polling_dispatch(struct stellar_polling_manager *polling_mgr);
 
 #ifdef __cplusplus
diff --git a/infra/stellar_core.c b/infra/stellar_core.c
index 49a3546..8574fd1 100644
--- a/infra/stellar_core.c
+++ b/infra/stellar_core.c
@@ -7,6 +7,7 @@
 #include "packet_manager_internal.h"
 #include "stellar/stellar.h"
 #include "stellar/module_manager.h"
+#include "polling_manager_internal.h"
 
 #define CORE_LOG_FATAL(format, ...) STELLAR_LOG_FATAL(__thread_local_logger, "core", format, ##__VA_ARGS__)
 #define CORE_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "core", format, ##__VA_ARGS__)
@@ -51,6 +52,9 @@ static void *worker_thread(void *arg)
     struct packet_manager *pkt_mgr = st->pkt_mgr;
     struct stellar_module_manager *mod_mgr = st->mod_mgr;
     struct mq_runtime *mq_rt = mq_runtime_new(st->mq_schema);
+    struct stellar_polling_manager *polling_mgr=stellar_module_get_polling_manager(mod_mgr);
+
+
 
     snprintf(thread_name, sizeof(thread_name), "stellar:%d", thread_id);
     prctl(PR_SET_NAME, (unsigned long long)thread_name, NULL, NULL, NULL);
@@ -102,12 +106,11 @@ static void *worker_thread(void *arg)
             {
                 packet_io_egress(pkt_io, thread_id, pkt, 1);
             }
-
-            // TODO polling
+            stellar_polling_dispatch(polling_mgr);
         }
 
     idle_tasks:
-        // TODO polling
+            stellar_polling_dispatch(polling_mgr);
 
         if (nr_pkt_rcv == 0)
         {