✨ feat(stellar core): module & node register in stellar_new

3 files changed, 69 insertions, 6 deletions

diff --git a/infra/session_manager/session_manager.c b/infra/session_manager/session_manager.c
index 02da2a6..1b9e5f4 100644
--- a/infra/session_manager/session_manager.c
+++ b/infra/session_manager/session_manager.c
@@ -303,7 +303,7 @@ struct session_manager *session_manager_new(struct packet_manager *pkt_mgr, cons
           SESSION_MANAGER_LOG_ERROR("failed to enable auto output");
           goto error_out;
      }
-
+#if 0
      if (packet_manager_register_node(pkt_mgr, "session_manager", PACKET_STAGE_FORWARD, PKT_TAG_KEY_IPPROTO, PKT_TAG_VAL_IPPROTO_TCP | PKT_TAG_VAL_IPPROTO_UDP, session_manager_on_packet_forward, sess_mgr))
      {
           SESSION_MANAGER_LOG_ERROR("failed to subscribe PACKET_STAGE_FORWARD");
@@ -314,6 +314,7 @@ struct session_manager *session_manager_new(struct packet_manager *pkt_mgr, cons
           SESSION_MANAGER_LOG_ERROR("failed to subscribe PACKET_STAGE_OUTPUT");
           goto error_out;
      }
+#endif
 
      sess_mgr->ex_sche = exdata_schema_new();
      if (sess_mgr->ex_sche == NULL)
diff --git a/infra/stellar_core.c b/infra/stellar_core.c
index 9141daf..52d9501 100644
--- a/infra/stellar_core.c
+++ b/infra/stellar_core.c
@@ -10,7 +10,6 @@
 #include "utils_internal.h"
 #include "packet_internal.h"
 #include "packet_manager.h"
-#include "module_manager_interna.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__)
@@ -131,6 +130,64 @@ static void stellar_thread_join(struct stellar *st)
     }
 }
 
+#include "stellar/monitor.h"
+#include "stellar/session.h"
+
+struct module_hooks mod_hooks[] = {
+                                {monitor_on_init, monitor_on_exit, NULL, NULL},
+                                {packet_manager_on_init, packet_manager_on_exit, packet_manager_on_thread_init, packet_manager_on_thread_exit},
+                                {session_manager_on_init, session_manager_on_exit, session_manager_on_thread_init, session_manager_on_thread_exit},
+                                {session_monitor_on_init, session_monitor_on_exit, NULL, NULL},
+                                };
+
+
+struct packet_node_spec
+{
+    const char *module_name;
+    const char *node_name;
+    enum packet_stage stage;
+    uint64_t interested_tag_key_bits;
+    uint64_t interested_tag_val_bits;
+    on_packet_callback *cb;
+};
+
+
+struct packet_node_spec packet_nodes[] = {
+    {SESSION_MANAGER_MODULE_NAME, "session_manager",PACKET_STAGE_FORWARD, PKT_TAG_KEY_IPPROTO, PKT_TAG_VAL_IPPROTO_TCP | PKT_TAG_VAL_IPPROTO_UDP, session_manager_on_packet_forward},
+    {SESSION_MANAGER_MODULE_NAME, "session_manager",PACKET_STAGE_OUTPUT, PKT_TAG_KEY_IPPROTO, PKT_TAG_VAL_IPPROTO_TCP | PKT_TAG_VAL_IPPROTO_UDP, session_manager_on_packet_output}
+};
+
+
+int module_register_packet_node(struct module_manager *mod_mgr, struct packet_node_spec *specs, size_t n_specs)
+{
+    struct module *pkt_mgr_mod = module_manager_get_module(mod_mgr, PACKET_MANAGER_MODULE_NAME);
+    struct packet_manager *pkt_mgr = module_to_packet_manager(pkt_mgr_mod);   
+
+    struct module *mod = NULL;
+
+    for(size_t i=0; i < n_specs; i++)
+    {
+        mod= module_manager_get_module(mod_mgr, specs[i].module_name);
+        if(mod == NULL)
+        {
+            CORE_LOG_ERROR("unable to get module %s", specs[i].module_name);
+            return -1;
+        }
+        if(packet_manager_register_node(pkt_mgr, 
+                                specs[i].node_name, 
+                                specs[i].stage, 
+                                specs[i].interested_tag_key_bits, 
+                                specs[i].interested_tag_val_bits, 
+                                specs[i].cb, 
+                                mod)<0)
+        {
+            CORE_LOG_ERROR("failed to subscribe PACKET_STAGE_FORWARD");
+            return -1;
+        }
+    }
+    return 0;
+}
+
 struct stellar *stellar_new(const char *toml_file)
 {
     if (toml_file == NULL)
@@ -168,7 +225,7 @@ struct stellar *stellar_new(const char *toml_file)
         goto error_out;
     }
 
-    st->mod_mgr = module_manager_new_with_toml(toml_file, st->thread_num, st->mq_schema, st->logger);
+    st->mod_mgr = module_manager_new(mod_hooks, count_of(mod_hooks),st->thread_num, toml_file, st->mq_schema, st->logger);
     if (st->mod_mgr == NULL)
     {
         CORE_LOG_ERROR("unable to create packet manager");
@@ -182,6 +239,12 @@ struct stellar *stellar_new(const char *toml_file)
         goto error_out;
     }
 
+    if(module_register_packet_node(st->mod_mgr, packet_nodes, count_of(packet_nodes)) != 0)
+    {
+        CORE_LOG_ERROR("unable to register packet node");
+        goto error_out;
+    }
+
     return st;
 
 error_out:
diff --git a/infra/utils_internal.h b/infra/utils_internal.h
index b2deb07..11c1f2a 100644
--- a/infra/utils_internal.h
+++ b/infra/utils_internal.h
@@ -12,6 +12,8 @@ extern "C"
 #include <stdlib.h>
 #include <string.h>
 
+#include "stellar/utils.h"
+
 #include "toml.h"
 #include "log_internal.h"
 
@@ -25,10 +27,7 @@ extern "C"
 #define ATOMIC_ZERO(x) __atomic_fetch_and(x, 0, __ATOMIC_RELAXED)
 #define ATOMIC_ADD(x, y) __atomic_fetch_add(x, y, __ATOMIC_RELAXED)
 #define ATOMIC_SET(x, y) __atomic_store_n(x, y, __ATOMIC_RELAXED)
-#define MIN(x, y) ((x) < (y) ? (x) : (y))
 
-#define likely(expr) __builtin_expect((expr), 1)
-#define unlikely(expr) __builtin_expect((expr), 0)
 
 /*
  * The maximum number of seconds that can be stored in the time_t value is 2147483647 –- a little over 68 years.