Optimize integration testing

- Add injection package plug-in
- Add libstellar_dynamic.so to facilitate unit testing of upper-level plug-ins

include/CMakeLists.txt

@@ -1,7 +1,7 @@
-install(FILES stellar/utils.h DESTINATION include/stellar/ COMPONENT Profile)
-install(FILES stellar/tuple.h DESTINATION include/stellar/ COMPONENT Profile)
-install(FILES stellar/packet.h DESTINATION include/stellar/ COMPONENT Profile)
-install(FILES stellar/session.h DESTINATION include/stellar/ COMPONENT Profile)
-install(FILES stellar/stellar.h DESTINATION include/stellar/ COMPONENT Profile)
-install(FILES stellar/session_mq.h DESTINATION include/stellar/ COMPONENT Profile)
-install(FILES stellar/session_exdata.h DESTINATION include/stellar/ COMPONENT Profile)
+install(FILES stellar/utils.h DESTINATION include/stellar/ COMPONENT HEADER)
+install(FILES stellar/tuple.h DESTINATION include/stellar/ COMPONENT HEADER)
+install(FILES stellar/packet.h DESTINATION include/stellar/ COMPONENT HEADER)
+install(FILES stellar/session.h DESTINATION include/stellar/ COMPONENT HEADER)
+install(FILES stellar/stellar.h DESTINATION include/stellar/ COMPONENT HEADER)
+install(FILES stellar/session_mq.h DESTINATION include/stellar/ COMPONENT HEADER)
+install(FILES stellar/session_exdata.h DESTINATION include/stellar/ COMPONENT HEADER)

include/stellar/stellar.h

@@ -52,10 +52,12 @@ int stellar_polling_plugin_register(struct stellar *st,  plugin_on_polling_func
 uint16_t stellar_get_current_thread_index();
 
 // return inject packet length, return 0 if failed
-int stellar_inject_tcp_rst(const struct session *sess, enum flow_direction inject_dir);
-int stellar_inject_tcp_fin(const struct session *sess, enum flow_direction inject_dir);
-int stellar_inject_payload(const struct session *sess, enum flow_direction inject_dir, const char *payload, uint16_t len);
-int stellar_inject_ctrl_msg(const struct session *sess, const struct sid_list *sids, const char *msg, uint16_t len);
+int stellar_inject_tcp_rst(struct stellar *st, const struct session *sess, enum flow_direction inject_dir);
+int stellar_inject_tcp_fin(struct stellar *st, const struct session *sess, enum flow_direction inject_dir);
+int stellar_inject_payload(struct stellar *st, const struct session *sess, enum flow_direction inject_dir, const char *payload, uint16_t len);
+int stellar_inject_ctrl_msg(struct stellar *st, const struct session *sess, const struct sid_list *sids, const char *msg, uint16_t len);
+
+int stellar_main(int argc, char **argv);
 
 #ifdef __cplusplus
 }

src/id_generator/CMakeLists.txt

@@ -1,4 +1,4 @@
 add_library(id_generator id_generator.cpp)
 target_include_directories(id_generator PUBLIC ${CMAKE_CURRENT_LIST_DIR})
 target_include_directories(id_generator PUBLIC ${CMAKE_SOURCE_DIR}/src/stellar)
-target_link_libraries(id_generator log core)
+target_link_libraries(id_generator log stellar_core)

src/plugin/CMakeLists.txt

@@ -1,4 +1,4 @@
 add_library(plugin_manager plugin_manager.cpp)
 target_include_directories(plugin_manager PUBLIC ${CMAKE_CURRENT_LIST_DIR})
 target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/)
-target_link_libraries(plugin_manager bitmap toml session_manager core ${CMAKE_DL_LIBS})
+target_link_libraries(plugin_manager bitmap toml session_manager stellar_core ${CMAKE_DL_LIBS})

src/plugin/plugin_manager.cpp

@@ -2,10 +2,6 @@
 #include "plugin_manager.h"
 #include "session_priv.h"
 #include "stellar_priv.h"
-
-
-#include "plugin_manager.h"
-
 #include "stellar/utils.h"
 #include "stellar/session.h"
 #include "stellar/session_exdata.h"
@@ -156,18 +152,6 @@ inline static struct session *plugin_manager_scratch_session_get()
 	return per_thread_scratch_sess;
 }
 
-inline struct plugin_manager_schema * stellar_plugin_manager_schema_get(struct stellar *st)
-{
-	return st->st_rt->plug_mgr;
-}
-
-inline int stellar_plugin_manager_schema_set(struct stellar *st, struct plugin_manager_schema *pm)
-{
-	if(st->st_rt->plug_mgr)return -1;
-	st->st_rt->plug_mgr=pm;
-	return 0;
-}
-
 UT_icd plugin_specs_icd = {sizeof(struct plugin_specific), NULL, NULL, NULL};
 
 static struct plugin_specific *plugin_specs_load(const char *toml_conf_path, int *spec_num)
@@ -251,9 +235,7 @@ struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char
 	}
 
 	pm->st = st;
-	stellar_plugin_manager_schema_set(st, pm);
-
-
+	stellar_set_plugin_manger(st, pm);
 	pm->tcp_topic_id=stellar_session_mq_create_topic(st, TOPIC_TCP, NULL, NULL);
 	pm->tcp_stream_topic_id=stellar_session_mq_create_topic(st, TOPIC_TCP_STREAM, tcp_stream_msg_free_fn, NULL);
 	pm->udp_topic_id=stellar_session_mq_create_topic(st, TOPIC_UDP, NULL, NULL);
@@ -334,7 +316,7 @@ UT_icd session_exdata_meta_icd = {sizeof(struct session_exdata_schema), NULL, se
 
 int stellar_session_exdata_new_index(struct stellar *st, const char *name, session_exdata_free *free_func,void *free_arg) 
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->session_exdata_schema_array == NULL)
 	{
 		utarray_new(plug_mgr->session_exdata_schema_array, &session_exdata_meta_icd);
@@ -423,7 +405,7 @@ void session_mq_free(struct session_message *head)
 
 int stellar_session_mq_get_topic_id(struct stellar *st, const char *topic_name)
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);;
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);;
 	if(topic_name == NULL || plug_mgr == NULL || plug_mgr->session_mq_schema_array == NULL)return -1;
 	unsigned int len = utarray_len(plug_mgr->session_mq_schema_array);
 	struct session_mq_topic_schema *t_schema;
@@ -440,7 +422,7 @@ int stellar_session_mq_get_topic_id(struct stellar *st, const char *topic_name)
 
 int stellar_session_mq_update_topic(struct stellar *st, int topic_id, msg_free_cb_func *msg_free_cb, void *msg_free_arg)
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->session_mq_schema_array == NULL)return -1;
 	unsigned int len = utarray_len(plug_mgr->session_mq_schema_array);
 	if(len < (unsigned int)topic_id)return -1;
@@ -453,7 +435,7 @@ int stellar_session_mq_update_topic(struct stellar *st, int topic_id, msg_free_c
 
 int stellar_session_mq_create_topic(struct stellar *st, const char *topic_name, msg_free_cb_func *msg_free_cb, void *msg_free_arg)
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->session_mq_schema_array == NULL)
 	{
 		utarray_new(plug_mgr->session_mq_schema_array, &session_mq_topic_schema_icd);
@@ -478,7 +460,7 @@ int stellar_session_mq_create_topic(struct stellar *st, const char *topic_name,
 
 int stellar_session_mq_destroy_topic(struct stellar *st, int topic_id)
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->session_mq_schema_array==NULL)return 0;
     unsigned int len = utarray_len(plug_mgr->session_mq_schema_array);
     if (len <= (unsigned int)topic_id)
@@ -559,7 +541,7 @@ UT_icd session_mq_subscriber_info_icd = {sizeof(struct session_mq_subscriber_inf
 int stellar_session_mq_subscribe(struct stellar *st, int topic_id, on_msg_cb_func *plugin_on_msg_cb, int plugin_id)
 {
 	if(plugin_id >= PACKET_PULGIN_ID_BASE)return -1;// ignore packet plugin
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->session_mq_schema_array==NULL)return -1;
     unsigned int len = utarray_len(plug_mgr->session_mq_schema_array);
     if (len <= (unsigned int)topic_id)return -1;
@@ -743,7 +725,7 @@ UT_icd registered_packet_plugin_array_icd = {sizeof(struct registered_packet_plu
 
 int stellar_packet_plugin_register(struct stellar *st, unsigned char ip_protocol, plugin_on_packet_func on_packet, void *plugin_env)
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->registered_packet_plugin_array == NULL)
 	{
 		utarray_new(plug_mgr->registered_packet_plugin_array, &registered_packet_plugin_array_icd);
@@ -783,7 +765,7 @@ UT_icd registered_polling_plugin_array_icd = {sizeof(struct registered_polling_p
 
 int stellar_polling_plugin_register(struct stellar *st, plugin_on_polling_func on_polling, void *plugin_env)
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->registered_polling_plugin_array == NULL)
 	{
 		utarray_new(plug_mgr->registered_polling_plugin_array, &registered_polling_plugin_array_icd);
@@ -828,7 +810,7 @@ int stellar_session_plugin_register(struct stellar *st,
 									session_ctx_free_func session_ctx_free, 
 									void *plugin_env)
 {
-	struct plugin_manager_schema *plug_mgr = stellar_plugin_manager_schema_get(st);
+	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
 	if(plug_mgr->registered_session_plugin_array == NULL)
 	{
 		utarray_new(plug_mgr->registered_session_plugin_array, &registered_session_plugin_schema_icd);

src/stellar/CMakeLists.txt

@@ -1,7 +1,17 @@
-add_library(core config.cpp stat.cpp stellar.cpp inject.cpp)
-target_link_libraries(core times plugin_manager session_manager ip_reassembly packet_io pthread fieldstat4 toml)
+set(SOURCE config.cpp stat.cpp stellar.cpp inject.cpp)
+set(LIBRARY times plugin_manager session_manager ip_reassembly packet_io pthread fieldstat4 toml)
+
+add_library(stellar_core STATIC ${SOURCE})
+target_link_libraries(stellar_core ${LIBRARY})
+
+add_library(stellar_devel SHARED ${SOURCE})
+target_link_libraries(stellar_devel ${LIBRARY})
+set_target_properties(stellar_devel PROPERTIES LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_LIST_DIR}/version.map")
 
 add_executable(stellar main.cpp)
-target_link_libraries(stellar core)
+target_link_libraries(stellar stellar_core)
 target_link_libraries(stellar "-rdynamic")
+set_target_properties(stellar PROPERTIES LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_LIST_DIR}/version.map")
+
 install(TARGETS stellar RUNTIME DESTINATION bin COMPONENT Program)
+install(TARGETS stellar_devel LIBRARY DESTINATION lib COMPONENT LIBRARIES)

src/stellar/inject.cpp

@@ -386,7 +386,7 @@ int build_udp_packet(const struct packet *first, const char *udp_pld, int pld_le
     return len - trim;
 }
 
-int inject_tcp_packet(const struct session *sess, enum flow_direction inject_dir, uint8_t tcp_flags, const char *payload, uint16_t len)
+int inject_tcp_packet(struct stellar *st, const struct session *sess, enum flow_direction inject_dir, uint8_t tcp_flags, const char *payload, uint16_t len)
 {
 #define TCP_FLAGS_LOG_FORMAT "URG:%d, ACK:%d, PSH:%d, RST:%d, SYN:%d, FIN:%d"
 #define TCP_FLAGS_LOG_VALUE(flags)                                   \
@@ -396,6 +396,8 @@ int inject_tcp_packet(const struct session *sess, enum flow_direction inject_dir
 
     uint16_t thr_idx = stellar_get_current_thread_index();
     uint64_t time_ms = stellar_get_monotonic_time_msec();
+    struct packet_io *packet_io = stellar_get_packet_io(st);
+    struct session_manager *sess_mgr = stellar_get_session_manager(st);
 
     if (session_get_type(sess) != SESSION_TYPE_TCP)
     {
@@ -442,8 +444,8 @@ int inject_tcp_packet(const struct session *sess, enum flow_direction inject_dir
     packet_parse(&inj_pkt, buff, pkt_len);
     packet_set_origin(&inj_pkt, PACKET_ORIGIN_USERSTACK);
     packet_set_origin_ctx(&inj_pkt, &meta);
-    session_manager_record_duplicated_packet(runtime->threads[thr_idx].sess_mgr, &inj_pkt, time_ms);
-    if (packet_io_inject(runtime->packet_io, thr_idx, &inj_pkt, 1) == 1)
+    session_manager_record_duplicated_packet(sess_mgr, &inj_pkt, time_ms);
+    if (packet_io_inject(packet_io, thr_idx, &inj_pkt, 1) == 1)
     {
         session_inc_stat((struct session *)sess, inject_dir, STAT_INJECTED_PACKETS_SUCCESS, 1);
         session_inc_stat((struct session *)sess, inject_dir, STAT_INJECTED_BYTES_SUCCESS, pkt_len);
@@ -462,10 +464,12 @@ int inject_tcp_packet(const struct session *sess, enum flow_direction inject_dir
     }
 }
 
-int inject_udp_packet(const struct session *sess, enum flow_direction inject_dir, const char *payload, uint16_t len)
+int inject_udp_packet(struct stellar *st, const struct session *sess, enum flow_direction inject_dir, const char *payload, uint16_t len)
 {
     uint16_t thr_idx = stellar_get_current_thread_index();
     uint64_t time_ms = stellar_get_monotonic_time_msec();
+    struct packet_io *packet_io = stellar_get_packet_io(st);
+    struct session_manager *sess_mgr = stellar_get_session_manager(st);
 
     if (session_get_type(sess) != SESSION_TYPE_UDP)
     {
@@ -505,8 +509,8 @@ int inject_udp_packet(const struct session *sess, enum flow_direction inject_dir
     packet_parse(&inj_pkt, buff, pkt_len);
     packet_set_origin(&inj_pkt, PACKET_ORIGIN_USERSTACK);
     packet_set_origin_ctx(&inj_pkt, &meta);
-    session_manager_record_duplicated_packet(runtime->threads[thr_idx].sess_mgr, &inj_pkt, time_ms);
-    if (packet_io_inject(runtime->packet_io, thr_idx, &inj_pkt, 1) == 1)
+    session_manager_record_duplicated_packet(sess_mgr, &inj_pkt, time_ms);
+    if (packet_io_inject(packet_io, thr_idx, &inj_pkt, 1) == 1)
     {
         session_inc_stat((struct session *)sess, inject_dir, STAT_INJECTED_PACKETS_SUCCESS, 1);
         session_inc_stat((struct session *)sess, inject_dir, STAT_INJECTED_BYTES_SUCCESS, pkt_len);
@@ -529,30 +533,30 @@ int inject_udp_packet(const struct session *sess, enum flow_direction inject_dir
  * Public API
  ******************************************************************************/
 
-int stellar_inject_tcp_rst(const struct session *sess, enum flow_direction inject_dir)
+int stellar_inject_tcp_rst(struct stellar *st, const struct session *sess, enum flow_direction inject_dir)
 {
-    return inject_tcp_packet(sess, inject_dir, TH_RST | TH_ACK, NULL, 0);
+    return inject_tcp_packet(st, sess, inject_dir, TH_RST | TH_ACK, NULL, 0);
 }
 
-int stellar_inject_tcp_fin(const struct session *sess, enum flow_direction inject_dir)
+int stellar_inject_tcp_fin(struct stellar *st, const struct session *sess, enum flow_direction inject_dir)
 {
-    return inject_tcp_packet(sess, inject_dir, TH_FIN | TH_ACK, NULL, 0);
+    return inject_tcp_packet(st, sess, inject_dir, TH_FIN | TH_ACK, NULL, 0);
 }
 
-int stellar_inject_payload(const struct session *sess, enum flow_direction inject_dir, const char *payload, uint16_t len)
+int stellar_inject_payload(struct stellar *st, const struct session *sess, enum flow_direction inject_dir, const char *payload, uint16_t len)
 {
     switch (session_get_type(sess))
     {
     case SESSION_TYPE_TCP:
-        return inject_tcp_packet(sess, inject_dir, TH_ACK, payload, len);
+        return inject_tcp_packet(st, sess, inject_dir, TH_ACK, payload, len);
     case SESSION_TYPE_UDP:
-        return inject_udp_packet(sess, inject_dir, payload, len);
+        return inject_udp_packet(st, sess, inject_dir, payload, len);
     default:
         return 0;
     }
 }
 
-int stellar_inject_ctrl_msg(const struct session *sess, const struct sid_list *sids, const char *msg, uint16_t len)
+int stellar_inject_ctrl_msg(struct stellar *st, const struct session *sess, const struct sid_list *sids, const char *msg, uint16_t len)
 {
     // TODO
     return 0;

src/stellar/main.cpp

@@ -1,169 +1,6 @@
-#include <stdio.h>
-#include <assert.h>
-#include <errno.h>
-#include <unistd.h>
-#include <string.h>
-#include <signal.h>
-#include <stdlib.h>
-
-#include "logo.h"
-#include "times.h"
-#include "config.h"
-#include "id_generator.h"
 #include "stellar_priv.h"
 
-static const char *log_config_file = "./conf/log.toml";
-static const char *stellar_config_file = "./conf/stellar.toml";
-
-static void signal_handler(int signo)
-{
-    if (signo == SIGINT)
-    {
-        STELLAR_LOG_STATE("SIGINT received, notify threads to exit !!!");
-        ATOMIC_SET(&runtime->need_exit, 1);
-    }
-
-    if (signo == SIGQUIT)
-    {
-        STELLAR_LOG_STATE("SIGQUIT received, notify threads to exit !!!");
-        ATOMIC_SET(&runtime->need_exit, 1);
-    }
-
-    if (signo == SIGTERM)
-    {
-        STELLAR_LOG_STATE("SIGTERM received, notify threads to exit !!!");
-        ATOMIC_SET(&runtime->need_exit, 1);
-    }
-
-    if (signo == SIGHUP)
-    {
-        STELLAR_LOG_STATE("SIGHUP received, reload log level !!!");
-        log_reload_level(log_config_file);
-    }
-}
-
-static int all_session_have_freed(void)
-{
-    for (int i = 0; i < config->io_opts.nr_threads; i++)
-    {
-        struct session_manager *sess_mgr = runtime->threads[i].sess_mgr;
-        struct session_manager_stat *sess_stat = session_manager_stat(sess_mgr);
-
-        if (ATOMIC_READ(&sess_stat->curr_nr_tcp_sess_used) != 0)
-        {
-            return 0;
-        }
-
-        if (ATOMIC_READ(&sess_stat->curr_nr_udp_sess_used) != 0)
-        {
-            return 0;
-        }
-    }
-
-    return 1;
-}
-
-static int all_stat_have_output(void)
-{
-    static int count = 0;
-    if (runtime->stat_last_output_ts == stellar_get_monotonic_time_msec())
-    {
-        count++;
-    }
-
-    return count == 2;
-}
-
 int main(int argc, char **argv)
 {
-    struct stellar st = {runtime};
-    stellar_update_time_cache();
-
-    signal(SIGINT, signal_handler);
-    signal(SIGQUIT, signal_handler);
-    signal(SIGTERM, signal_handler);
-    signal(SIGHUP, signal_handler);
-
-    if (log_init(log_config_file) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to init log");
-        goto error_out;
-    }
-    STELLAR_LOG_STATE("start stellar (version: %s)\n %s", __stellar_version, logo_str);
-
-    if (stellar_load_config(stellar_config_file, config) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to load config file");
-        goto error_out;
-    }
-    stellar_print_config(config);
-    STELLAR_LOG_DEBUG("sizeof(struct session) = %lu bytes", sizeof(struct session));
-
-    if (id_generator_init(config->dev_opts.base, config->dev_opts.offset) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to init id generator");
-        goto error_out;
-    }
-
-    runtime->stat = stellar_stat_new(config->io_opts.nr_threads);
-    if (runtime->stat == NULL)
-    {
-        STELLAR_LOG_ERROR("unable to create stellar stat");
-        goto error_out;
-    }
-    runtime->plug_mgr = plugin_manager_init(&st, "./stellar_plugin/spec.toml");
-    if (runtime->plug_mgr == NULL)
-    {
-        STELLAR_LOG_ERROR("unable to create plugin manager");
-        goto error_out;
-    }
-
-    runtime->packet_io = packet_io_new(&config->io_opts);
-    if (runtime->packet_io == NULL)
-    {
-        STELLAR_LOG_ERROR("unable to create packet io");
-        goto error_out;
-    }
-
-    if (stellar_thread_init(runtime, config) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to init thread context");
-        goto error_out;
-    }
-
-    if (stellar_thread_run(runtime, config) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to create worker thread");
-        goto error_out;
-    }
-
-    runtime->stat_last_output_ts = stellar_get_monotonic_time_msec();
-    while (!ATOMIC_READ(&runtime->need_exit))
-    {
-        stellar_update_time_cache();
-        if (stellar_get_monotonic_time_msec() - runtime->stat_last_output_ts > 2000)
-        {
-            runtime->stat_last_output_ts = stellar_get_monotonic_time_msec();
-            stellar_stat_output(runtime->stat);
-        }
-        usleep(1000); // 1ms
-
-        // Only available in dump file mode, automatically exits when all sessions have been released
-        if (packet_io_wait_exit(runtime->packet_io) && all_session_have_freed() && all_stat_have_output())
-        {
-            STELLAR_LOG_STATE("all sessions have been released and all stat have been output, notify threads to exit !!!");
-            ATOMIC_SET(&runtime->need_exit, 1);
-        }
-    }
-
-error_out:
-    stellar_thread_join(runtime, config);
-    stellar_thread_clean(runtime, config);
-    packet_io_free(runtime->packet_io);
-    plugin_manager_exit(runtime->plug_mgr);
-    stellar_stat_free(runtime->stat);
-    STELLAR_LOG_STATE("stellar exit !!!\n");
-    log_free();
-
-    return 0;
+   return stellar_main(argc, argv);
 }

src/stellar/stellar.cpp

@@ -2,17 +2,53 @@
 #include <assert.h>
 #include <unistd.h>
 #include <pthread.h>
+#include <signal.h>
 #include <sys/prctl.h>
 
+#include "stat.h"
+#include "logo.h"
 #include "times.h"
 #include "config.h"
+#include "id_generator.h"
 #include "stellar_priv.h"
+#include "plugin_manager.h"
 
-struct stellar_runtime __runtime = {0};
-struct stellar_runtime *runtime = &__runtime;
+struct stellar_thread
+{
+    pthread_t tid;
+    uint16_t idx;
+    uint64_t is_runing;
+    uint64_t timing_wheel_last_update_ts;
+    struct ip_reassembly *ip_mgr;
+    struct session_manager *sess_mgr;
+    struct stellar_runtime *runtime;
+};
 
-struct stellar_config __config = {0};
-struct stellar_config *config = &__config;
+struct stellar_runtime
+{
+    uint64_t need_exit;
+    uint64_t stat_last_output_ts;
+    struct stellar_stat *stat;
+    struct packet_io *packet_io;
+    struct plugin_manager_schema *plug_mgr;
+    struct stellar_thread threads[MAX_THREAD_NUM];
+};
+
+struct stellar
+{
+    struct stellar_runtime runtime;
+    struct stellar_config config;
+};
+
+static uint64_t need_exit = 0;
+static thread_local uint16_t __thread_id = 0;
+static const char *log_config_file = "./conf/log.toml";
+static const char *main_config_file = "./conf/stellar.toml";
+static const char *plugin_config_file = "./stellar_plugin/spec.toml";
+
+/******************************************************************************
+ * Stellar Thread Main Loop
+ ******************************************************************************/
 
 static void update_session_stat(struct session *sess, struct packet *pkt)
 {
@@ -83,6 +119,7 @@ static inline void free_expired_sessions(struct session_manager *sess_mgr, uint6
 
 static inline void merge_thread_stat(struct stellar_thread *thread, uint64_t now)
 {
+    struct stellar_runtime *runtime = thread->runtime;
     struct thread_stat thr_stat = {
         packet_io_stat(runtime->packet_io, thread->idx),
         ip_reassembly_stat(thread->ip_mgr),
@@ -101,13 +138,15 @@ static void *work_thread(void *arg)
     struct packet *defraged_pkt = NULL;
     struct packet packets[RX_BURST_MAX];
     struct session *sess = NULL;
-    struct packet_io *packet_io = runtime->packet_io;
-    struct plugin_manager_schema *plug_mgr = runtime->plug_mgr;
     struct stellar_thread *thread = (struct stellar_thread *)arg;
     struct ip_reassembly *ip_reass = thread->ip_mgr;
     struct session_manager *sess_mgr = thread->sess_mgr;
+    struct stellar_runtime *runtime = thread->runtime;
+    struct packet_io *packet_io = runtime->packet_io;
+    struct plugin_manager_schema *plug_mgr = runtime->plug_mgr;
+
     uint16_t thr_idx = thread->idx;
-    stellar_set_current_thread_index(thr_idx);
+    __thread_id = thr_idx;
 
     snprintf(thd_name, sizeof(thd_name), "stellar:%d", thr_idx);
     prctl(PR_SET_NAME, (unsigned long long)thd_name, NULL, NULL, NULL);
@@ -121,7 +160,7 @@ static void *work_thread(void *arg)
     ATOMIC_SET(&thread->is_runing, 1);
     STELLAR_LOG_STATE("worker thread %d runing", thr_idx);
 
-    while (ATOMIC_READ(&runtime->need_exit) == 0)
+    while (ATOMIC_READ(&need_exit) == 0)
     {
         now = stellar_get_monotonic_time_msec();
         memset(packets, 0, sizeof(packets));
@@ -236,19 +275,70 @@ static void *work_thread(void *arg)
     return NULL;
 }
 
-static thread_local uint16_t __thread_id = 0;
+/******************************************************************************
+ * Stellar Main Function
+ ******************************************************************************/
 
-uint16_t stellar_get_current_thread_index()
+static void signal_handler(int signo)
 {
-    return __thread_id;
+    if (signo == SIGINT)
+    {
+        STELLAR_LOG_STATE("SIGINT received, notify threads to exit !!!");
+        ATOMIC_SET(&need_exit, 1);
     }
 
-void stellar_set_current_thread_index(uint16_t idx)
+    if (signo == SIGQUIT)
     {
-    __thread_id = idx;
+        STELLAR_LOG_STATE("SIGQUIT received, notify threads to exit !!!");
+        ATOMIC_SET(&need_exit, 1);
     }
 
-int stellar_thread_init(struct stellar_runtime *runtime, struct stellar_config *config)
+    if (signo == SIGTERM)
+    {
+        STELLAR_LOG_STATE("SIGTERM received, notify threads to exit !!!");
+        ATOMIC_SET(&need_exit, 1);
+    }
+
+    if (signo == SIGHUP)
+    {
+        STELLAR_LOG_STATE("SIGHUP received, reload log level !!!");
+        log_reload_level(log_config_file);
+    }
+}
+
+static int all_session_have_freed(struct stellar_runtime *runtime, struct stellar_config *config)
+{
+    for (int i = 0; i < config->io_opts.nr_threads; i++)
+    {
+        struct session_manager *sess_mgr = runtime->threads[i].sess_mgr;
+        struct session_manager_stat *sess_stat = session_manager_stat(sess_mgr);
+
+        if (ATOMIC_READ(&sess_stat->curr_nr_tcp_sess_used) != 0)
+        {
+            return 0;
+        }
+
+        if (ATOMIC_READ(&sess_stat->curr_nr_udp_sess_used) != 0)
+        {
+            return 0;
+        }
+    }
+
+    return 1;
+}
+
+static int all_stat_have_output(struct stellar_runtime *runtime, struct stellar_config *config)
+{
+    static int count = 0;
+    if (runtime->stat_last_output_ts == stellar_get_monotonic_time_msec())
+    {
+        count++;
+    }
+
+    return count == 2;
+}
+
+static int stellar_thread_init(struct stellar_runtime *runtime, struct stellar_config *config)
 {
     uint64_t now = stellar_get_monotonic_time_msec();
     for (uint16_t i = 0; i < config->io_opts.nr_threads; i++)
@@ -269,12 +359,13 @@ int stellar_thread_init(struct stellar_runtime *runtime, struct stellar_config *
             STELLAR_LOG_ERROR("unable to create ip reassemble manager");
             return -1;
         }
+        thread->runtime = runtime;
     }
 
     return 0;
 }
 
-void stellar_thread_clean(struct stellar_runtime *runtime, struct stellar_config *config)
+static void stellar_thread_clean(struct stellar_runtime *runtime, struct stellar_config *config)
 {
     for (uint16_t i = 0; i < config->io_opts.nr_threads; i++)
     {
@@ -288,7 +379,7 @@ void stellar_thread_clean(struct stellar_runtime *runtime, struct stellar_config
     }
 }
 
-int stellar_thread_run(struct stellar_runtime *runtime, struct stellar_config *config)
+static int stellar_thread_run(struct stellar_runtime *runtime, struct stellar_config *config)
 {
     for (uint16_t i = 0; i < config->io_opts.nr_threads; i++)
     {
@@ -303,7 +394,7 @@ int stellar_thread_run(struct stellar_runtime *runtime, struct stellar_config *c
     return 0;
 }
 
-void stellar_thread_join(struct stellar_runtime *runtime, struct stellar_config *config)
+static void stellar_thread_join(struct stellar_runtime *runtime, struct stellar_config *config)
 {
     for (uint16_t i = 0; i < config->io_opts.nr_threads; i++)
     {
@@ -315,3 +406,133 @@ void stellar_thread_join(struct stellar_runtime *runtime, struct stellar_config
         }
     }
 }
+
+int stellar_main(int argc, char **argv)
+{
+    static struct stellar st = {0};
+    static struct stellar_runtime *runtime = &st.runtime;
+    static struct stellar_config *config = &st.config;
+
+    stellar_update_time_cache();
+
+    signal(SIGINT, signal_handler);
+    signal(SIGQUIT, signal_handler);
+    signal(SIGTERM, signal_handler);
+    signal(SIGHUP, signal_handler);
+
+    if (log_init(log_config_file) != 0)
+    {
+        STELLAR_LOG_ERROR("unable to init log");
+        goto error_out;
+    }
+    STELLAR_LOG_STATE("start stellar (version: %s)\n %s", __stellar_version, logo_str);
+    STELLAR_LOG_STATE("log config file    : %s", log_config_file);
+    STELLAR_LOG_STATE("main config file   : %s", main_config_file);
+    STELLAR_LOG_STATE("plugin config file : %s", plugin_config_file);
+
+    if (stellar_load_config(main_config_file, config) != 0)
+    {
+        STELLAR_LOG_ERROR("unable to load config file");
+        goto error_out;
+    }
+    stellar_print_config(config);
+    STELLAR_LOG_DEBUG("sizeof(struct session) = %lu bytes", sizeof(struct session));
+
+    if (id_generator_init(config->dev_opts.base, config->dev_opts.offset) != 0)
+    {
+        STELLAR_LOG_ERROR("unable to init id generator");
+        goto error_out;
+    }
+
+    runtime->stat = stellar_stat_new(config->io_opts.nr_threads);
+    if (runtime->stat == NULL)
+    {
+        STELLAR_LOG_ERROR("unable to create stellar stat");
+        goto error_out;
+    }
+    runtime->plug_mgr = plugin_manager_init(&st, plugin_config_file);
+    if (runtime->plug_mgr == NULL)
+    {
+        STELLAR_LOG_ERROR("unable to create plugin manager");
+        goto error_out;
+    }
+
+    runtime->packet_io = packet_io_new(&config->io_opts);
+    if (runtime->packet_io == NULL)
+    {
+        STELLAR_LOG_ERROR("unable to create packet io");
+        goto error_out;
+    }
+
+    if (stellar_thread_init(runtime, config) != 0)
+    {
+        STELLAR_LOG_ERROR("unable to init thread context");
+        goto error_out;
+    }
+
+    if (stellar_thread_run(runtime, config) != 0)
+    {
+        STELLAR_LOG_ERROR("unable to create worker thread");
+        goto error_out;
+    }
+
+    runtime->stat_last_output_ts = stellar_get_monotonic_time_msec();
+    while (!ATOMIC_READ(&need_exit))
+    {
+        stellar_update_time_cache();
+        if (stellar_get_monotonic_time_msec() - runtime->stat_last_output_ts > 2000)
+        {
+            runtime->stat_last_output_ts = stellar_get_monotonic_time_msec();
+            stellar_stat_output(runtime->stat);
+        }
+        usleep(1000); // 1ms
+
+        // Only available in dump file mode, automatically exits when all sessions have been released
+        if (packet_io_wait_exit(runtime->packet_io) && all_session_have_freed(runtime, config) && all_stat_have_output(runtime, config))
+        {
+            STELLAR_LOG_STATE("all sessions have been released and all stat have been output, notify threads to exit !!!");
+            ATOMIC_SET(&need_exit, 1);
+        }
+    }
+
+error_out:
+    stellar_thread_join(runtime, config);
+    stellar_thread_clean(runtime, config);
+    packet_io_free(runtime->packet_io);
+    plugin_manager_exit(runtime->plug_mgr);
+    stellar_stat_free(runtime->stat);
+    STELLAR_LOG_STATE("stellar exit !!!\n");
+    log_free();
+
+    return 0;
+}
+
+/******************************************************************************
+ * Stellar Utility Function
+ ******************************************************************************/
+
+struct packet_io *stellar_get_packet_io(const struct stellar *st)
+{
+    return st->runtime.packet_io;
+}
+
+struct plugin_manager_schema *stellar_get_plugin_manager(const struct stellar *st)
+{
+    return st->runtime.plug_mgr;
+}
+
+void stellar_set_plugin_manger(struct stellar *st, struct plugin_manager_schema *plug_mgr)
+{
+    st->runtime.plug_mgr = plug_mgr;
+}
+
+struct session_manager *stellar_get_session_manager(const struct stellar *st)
+{
+    uint16_t idx = stellar_get_current_thread_index();
+    return st->runtime.threads[idx].sess_mgr;
+}
+
+uint16_t stellar_get_current_thread_index()
+{
+    return __thread_id;
+}

src/stellar/stellar_priv.h

@@ -6,49 +6,16 @@ extern "C"
 {
 #endif
 
-#include "stat.h"
-#include "plugin_manager.h"
 #include "stellar/stellar.h"
 
 #define STELLAR_LOG_STATE(format, ...) LOG_STATE("stellar", format, ##__VA_ARGS__)
 #define STELLAR_LOG_ERROR(format, ...) LOG_ERROR("stellar", format, ##__VA_ARGS__)
 #define STELLAR_LOG_DEBUG(format, ...) LOG_DEBUG("stellar", format, ##__VA_ARGS__)
 
-struct stellar_thread
-{
-    pthread_t tid;
-    uint16_t idx;
-    uint64_t is_runing;
-    uint64_t timing_wheel_last_update_ts;
-    struct ip_reassembly *ip_mgr;
-    struct session_manager *sess_mgr;
-};
-
-struct stellar_runtime
-{
-    uint64_t need_exit;
-    uint64_t stat_last_output_ts;
-    struct stellar_stat *stat;
-    struct packet_io *packet_io;
-    struct plugin_manager_schema *plug_mgr;
-    struct stellar_thread threads[MAX_THREAD_NUM];
-};
-
-//FIXME rename stellar_runtime to stellar
-struct stellar
-{
-    struct stellar_runtime *st_rt;
-};
-
-extern struct stellar_runtime *runtime;
-extern struct stellar_config *config;
-
-void stellar_set_current_thread_index(uint16_t idx);
-
-int stellar_thread_init(struct stellar_runtime *runtime, struct stellar_config *config);
-void stellar_thread_clean(struct stellar_runtime *runtime, struct stellar_config *config);
-int stellar_thread_run(struct stellar_runtime *runtime, struct stellar_config *config);
-void stellar_thread_join(struct stellar_runtime *runtime, struct stellar_config *config);
+struct packet_io *stellar_get_packet_io(const struct stellar *st);
+struct session_manager *stellar_get_session_manager(const struct stellar *st);
+struct plugin_manager_schema *stellar_get_plugin_manager(const struct stellar *st);
+void stellar_set_plugin_manger(struct stellar *st, struct plugin_manager_schema *plug_mgr);
 
 #ifdef __cplusplus
 }

src/stellar/version.map

@@ -0,0 +1,84 @@
+LIBSTELLAR_DEVEL {
+global:
+	packet_get_innermost_tuple2;
+	packet_get_outermost_tuple2;
+	packet_get_innermost_tuple4;
+	packet_get_outermost_tuple4;
+	packet_get_innermost_tuple6;
+	packet_get_outermost_tuple6;
+	packet_get_innermost_layer;
+	packet_get_outermost_layer;
+	packet_get_direction;
+	packet_get_session_id;
+	packet_prepend_sid_list;
+	packet_get_layers_number;
+	packet_get_layer;
+	packet_get_data;
+	packet_get_len;
+	packet_get_payload;
+	packet_get_payload_len;
+	packet_set_action;
+	packet_get_action;
+
+	session_exdata_free;
+	stellar_session_exdata_new_index;
+	session_exdata_set;
+	session_exdata_get;
+
+	stellar_session_mq_create_topic;
+	stellar_session_mq_get_topic_id;
+	stellar_session_mq_update_topic;
+	stellar_session_mq_destroy_topic;
+	stellar_session_mq_subscribe;
+	session_mq_publish_message;
+	session_mq_ignore_message;
+	session_mq_unignore_message;
+
+	session_is_symmetric;
+	session_has_duplicate_traffic;
+	session_get_type;
+	session_get_current_state;
+	session_get0_current_packet;
+	session_get0_current_payload;
+	session_get_closing_reason;
+	session_get_direction;
+	session_get_current_flow_direction;
+	session_get_first_packet;
+	session_get_tuple6;
+	session_get_tuple6_direction;
+	session_get_id;
+	session_get_timestamp;
+	session_get_stat;
+	session_get0_readable_addr;
+	session_set_discard;
+
+	stellar_session_plugin_register;
+	stellar_session_plugin_dettach_current_session;
+	stellar_packet_plugin_register;
+	stellar_polling_plugin_register;
+	stellar_get_current_thread_index;
+	stellar_inject_tcp_rst;
+	stellar_inject_tcp_fin;
+	stellar_inject_payload;
+	stellar_inject_ctrl_msg;
+	stellar_main;
+
+	tuple2_hash;
+	tuple4_hash;
+	tuple5_hash;
+	tuple6_hash;
+	tuple2_cmp;
+	tuple4_cmp;
+	tuple5_cmp;
+	tuple6_cmp;
+	tuple2_reverse;
+	tuple4_reverse;
+	tuple5_reverse;
+	tuple6_reverse;
+	tuple2_to_str;
+	tuple4_to_str;
+	tuple5_to_str;
+	tuple6_to_str;
+
+local: *;
+};

test/CMakeLists.txt

@@ -1,42 +1,2 @@
-add_library(libpacket_injector packet_injector.cpp packet_injector_test_frame.cpp)
-target_include_directories(libpacket_injector PUBLIC ${CMAKE_CURRENT_LIST_DIR})
-target_link_libraries(libpacket_injector core gtest)
-
-add_executable(gtest_inject_tcp_rst_after_recv_syn_ack gtest_inject_tcp_rst_after_recv_syn_ack.cpp)
-target_link_libraries(gtest_inject_tcp_rst_after_recv_syn_ack libpacket_injector)
-
-add_executable(gtest_inject_tcp_rst_after_recv_sub_ack gtest_inject_tcp_rst_after_recv_sub_ack.cpp)
-target_link_libraries(gtest_inject_tcp_rst_after_recv_sub_ack libpacket_injector)
-
-add_executable(gtest_inject_tcp_rst_after_recv_c2s_first_payload gtest_inject_tcp_rst_after_recv_c2s_first_payload.cpp)
-target_link_libraries(gtest_inject_tcp_rst_after_recv_c2s_first_payload libpacket_injector)
-
-add_executable(gtest_inject_tcp_rst_after_recv_s2c_first_payload gtest_inject_tcp_rst_after_recv_s2c_first_payload.cpp)
-target_link_libraries(gtest_inject_tcp_rst_after_recv_s2c_first_payload libpacket_injector)
-
-add_executable(gtest_inject_tcp_payload_after_recv_c2s_first_payload gtest_inject_tcp_payload_after_recv_c2s_first_payload.cpp)
-target_link_libraries(gtest_inject_tcp_payload_after_recv_c2s_first_payload libpacket_injector)
-
-add_executable(gtest_inject_tcp_payload_after_recv_s2c_first_payload gtest_inject_tcp_payload_after_recv_s2c_first_payload.cpp)
-target_link_libraries(gtest_inject_tcp_payload_after_recv_s2c_first_payload libpacket_injector)
-
-add_executable(gtest_inject_tcp_payload_fin_rst_after_recv_c2s_first_payload gtest_inject_tcp_payload_fin_rst_after_recv_c2s_first_payload.cpp)
-target_link_libraries(gtest_inject_tcp_payload_fin_rst_after_recv_c2s_first_payload libpacket_injector)
-
-include(GoogleTest)
-gtest_discover_tests(gtest_inject_tcp_rst_after_recv_syn_ack)
-gtest_discover_tests(gtest_inject_tcp_rst_after_recv_sub_ack)
-gtest_discover_tests(gtest_inject_tcp_rst_after_recv_c2s_first_payload)
-gtest_discover_tests(gtest_inject_tcp_rst_after_recv_s2c_first_payload)
-gtest_discover_tests(gtest_inject_tcp_payload_after_recv_c2s_first_payload)
-gtest_discover_tests(gtest_inject_tcp_payload_after_recv_s2c_first_payload)
-gtest_discover_tests(gtest_inject_tcp_payload_fin_rst_after_recv_c2s_first_payload)
-
-add_executable(packet_injector packet_injector.cpp)
-target_link_libraries(packet_injector core gtest)
-
-add_executable(packet_parser packet_parser.cpp)
-target_link_libraries(packet_parser packet pcap)
-
-file(COPY ./conf/ DESTINATION ./conf/)
-file(COPY ./pcap/ DESTINATION ./pcap/)
+add_subdirectory(packet_inject)
+add_subdirectory(packet_parser)

test/packet_inject/CMakeLists.txt

@@ -0,0 +1,30 @@
+# build packet_injector
+add_executable(packet_injector packet_inject_main.cpp packet_inject_plugin.cpp)
+target_link_libraries(packet_injector "-rdynamic")
+target_link_libraries(packet_injector stellar_devel)
+
+# build libpacket_inject_plugin.so
+add_library(packet_inject_plugin SHARED packet_inject_plugin.cpp)
+target_include_directories(packet_inject_plugin PUBLIC ${CMAKE_SOURCE_DIR}/include/)
+set_target_properties(packet_inject_plugin PROPERTIES LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_LIST_DIR}/version.map")
+
+# build gtest
+function(packet_inject_add_case EXEC_NAME)
+    add_executable(${EXEC_NAME} ${EXEC_NAME}.cpp packet_inject_main.cpp packet_inject_plugin.cpp packet_inject_test.cpp)
+    target_link_libraries(${EXEC_NAME} "-rdynamic")
+    target_link_libraries(${EXEC_NAME} stellar_devel gtest)
+    gtest_discover_tests(${EXEC_NAME})
+endfunction()
+
+include(GoogleTest)
+packet_inject_add_case(gtest_inject_tcp_rst_after_recv_syn_ack)
+packet_inject_add_case(gtest_inject_tcp_rst_after_recv_sub_ack)
+packet_inject_add_case(gtest_inject_tcp_rst_after_recv_c2s_first_payload)
+packet_inject_add_case(gtest_inject_tcp_rst_after_recv_s2c_first_payload)
+packet_inject_add_case(gtest_inject_tcp_payload_after_recv_c2s_first_payload)
+packet_inject_add_case(gtest_inject_tcp_payload_after_recv_s2c_first_payload)
+packet_inject_add_case(gtest_inject_tcp_payload_fin_rst_after_recv_c2s_first_payload)
+
+file(COPY ./conf/ DESTINATION ./conf/)
+file(COPY ./pcap/ DESTINATION ./pcap/)
+file(COPY ./stellar_plugin/ DESTINATION ./stellar_plugin/)

test/packet_inject/gtest_inject_tcp_payload_after_recv_c2s_first_payload.cpp

@@ -1,6 +1,6 @@
 #include <gtest/gtest.h>
 
-#include "packet_injector_test_frame.h"
+#include "packet_inject_test.h"
 
 TEST(INJECT_IPV4_BASED_TCP_PAYLOAD, AFTER_RECV_C2S_FIRST_PAYLOAD)
 {
@@ -11,7 +11,7 @@ TEST(INJECT_IPV4_BASED_TCP_PAYLOAD, AFTER_RECV_C2S_FIRST_PAYLOAD)
     snprintf(work_dir, sizeof(work_dir), "%s/%s", current_dir, "inject_ipv4_based_tcp_payload_after_recv_c2s_first_payload");
     snprintf(input_dir, sizeof(input_dir), "%s/%s", current_dir, "pcap/inject_ipv4_based_tcp_payload_after_recv_c2s_first_payload/test/");
 
-    struct packet_injector_case test = {
+    struct packet_inject_case test = {
         // descriptor
         .finish_clean_work_dir = 0,
         .descriptor = "Inject IPv4 based TCP Payload after receiving C2S first payload packet.",
@@ -52,7 +52,7 @@ TEST(INJECT_IPV4_BASED_TCP_PAYLOAD, AFTER_RECV_C2S_FIRST_PAYLOAD)
         .diff_skip_pattern = "-I frame.time -I frame.time_epoch -I ip.id -I ip.ttl -I ip.checksum -I tcp.checksum -I tcp.window_size",
     };
 
-    packet_injector_test_frame_run(&test);
+    packet_inject_test(&test);
 }
 
 int main(int argc, char **argv)

test/packet_inject/gtest_inject_tcp_payload_after_recv_s2c_first_payload.cpp

@@ -1,6 +1,6 @@
 #include <gtest/gtest.h>
 
-#include "packet_injector_test_frame.h"
+#include "packet_inject_test.h"
 
 TEST(INJECT_IPV4_BASED_TCP_PAYLOAD, AFTER_RECV_S2C_FIRST_PAYLOAD)
 {
@@ -11,7 +11,7 @@ TEST(INJECT_IPV4_BASED_TCP_PAYLOAD, AFTER_RECV_S2C_FIRST_PAYLOAD)
     snprintf(work_dir, sizeof(work_dir), "%s/%s", current_dir, "inject_ipv4_based_tcp_payload_after_recv_s2c_first_payload");
     snprintf(input_dir, sizeof(input_dir), "%s/%s", current_dir, "pcap/inject_ipv4_based_tcp_payload_after_recv_s2c_first_payload/test/");
 
-    struct packet_injector_case test = {
+    struct packet_inject_case test = {
         // descriptor
         .finish_clean_work_dir = 0,
         .descriptor = "Inject IPv4 based TCP Payload after receiving S2C first payload packet.",
@@ -52,7 +52,7 @@ TEST(INJECT_IPV4_BASED_TCP_PAYLOAD, AFTER_RECV_S2C_FIRST_PAYLOAD)
         .diff_skip_pattern = "-I frame.time -I frame.time_epoch -I ip.id -I ip.ttl -I ip.checksum -I tcp.checksum -I tcp.window_size",
     };
 
-    packet_injector_test_frame_run(&test);
+    packet_inject_test(&test);
 }
 
 int main(int argc, char **argv)

test/packet_inject/gtest_inject_tcp_payload_fin_rst_after_recv_c2s_first_payload.cpp

@@ -1,6 +1,6 @@
 #include <gtest/gtest.h>
 
-#include "packet_injector_test_frame.h"
+#include "packet_inject_test.h"
 
 TEST(INJECT_IPV4_BASED_TCP_PAYLOAD_FIN_RST, AFTER_RECV_C2S_FIRST_PAYLOAD)
 {
@@ -11,7 +11,7 @@ TEST(INJECT_IPV4_BASED_TCP_PAYLOAD_FIN_RST, AFTER_RECV_C2S_FIRST_PAYLOAD)
     snprintf(work_dir, sizeof(work_dir), "%s/%s", current_dir, "inject_ipv4_based_tcp_payload_fin_rst_after_recv_c2s_first_payload");
     snprintf(input_dir, sizeof(input_dir), "%s/%s", current_dir, "pcap/inject_ipv4_based_tcp_payload_fin_rst_after_recv_c2s_first_payload/test/");
 
-    struct packet_injector_case test = {
+    struct packet_inject_case test = {
         // descriptor
         .finish_clean_work_dir = 0,
         .descriptor = "Inject IPv4 based TCP Payload & FIN & RST after receiving C2S first payload packet.",
@@ -60,7 +60,7 @@ TEST(INJECT_IPV4_BASED_TCP_PAYLOAD_FIN_RST, AFTER_RECV_C2S_FIRST_PAYLOAD)
         .diff_skip_pattern = "-I frame.time -I frame.time_epoch -I ip.id -I ip.ttl -I ip.checksum -I tcp.checksum -I tcp.window_size",
     };
 
-    packet_injector_test_frame_run(&test);
+    packet_inject_test(&test);
 }
 
 int main(int argc, char **argv)

test/packet_inject/gtest_inject_tcp_rst_after_recv_c2s_first_payload.cpp

@@ -1,6 +1,6 @@
 #include <gtest/gtest.h>
 
-#include "packet_injector_test_frame.h"
+#include "packet_inject_test.h"
 
 TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_C2S_FIRST_PAYLOAD)
 {
@@ -11,7 +11,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_C2S_FIRST_PAYLOAD)
     snprintf(work_dir, sizeof(work_dir), "%s/%s", current_dir, "inject_ipv4_based_tcp_rst_after_recv_c2s_first_payload");
     snprintf(input_dir, sizeof(input_dir), "%s/%s", current_dir, "pcap/inject_ipv4_based_tcp_rst_after_recv_c2s_first_payload/test/");
 
-    struct packet_injector_case test = {
+    struct packet_inject_case test = {
         // descriptor
         .finish_clean_work_dir = 0,
         .descriptor = "Inject IPv4 based TCP RST after receiving C2S first payload packet.",
@@ -44,7 +44,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_C2S_FIRST_PAYLOAD)
         .diff_skip_pattern = "-I frame.time -I frame.time_epoch -I ip.id -I ip.ttl -I ip.checksum -I tcp.checksum -I tcp.window_size",
     };
 
-    packet_injector_test_frame_run(&test);
+    packet_inject_test(&test);
 }
 
 int main(int argc, char **argv)

test/packet_inject/gtest_inject_tcp_rst_after_recv_s2c_first_payload.cpp

@@ -1,6 +1,6 @@
 #include <gtest/gtest.h>
 
-#include "packet_injector_test_frame.h"
+#include "packet_inject_test.h"
 
 TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_S2C_FIRST_PAYLOAD)
 {
@@ -11,7 +11,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_S2C_FIRST_PAYLOAD)
     snprintf(work_dir, sizeof(work_dir), "%s/%s", current_dir, "inject_ipv4_based_tcp_rst_after_recv_s2c_first_payload");
     snprintf(input_dir, sizeof(input_dir), "%s/%s", current_dir, "pcap/inject_ipv4_based_tcp_rst_after_recv_s2c_first_payload/test/");
 
-    struct packet_injector_case test = {
+    struct packet_inject_case test = {
         // descriptor
         .finish_clean_work_dir = 0,
         .descriptor = "Inject IPv4 based TCP RST after receiving S2C first payload packet.",
@@ -44,7 +44,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_S2C_FIRST_PAYLOAD)
         .diff_skip_pattern = "-I frame.time -I frame.time_epoch -I ip.id -I ip.ttl -I ip.checksum -I tcp.checksum -I tcp.window_size",
     };
 
-    packet_injector_test_frame_run(&test);
+    packet_inject_test(&test);
 }
 
 int main(int argc, char **argv)

test/packet_inject/gtest_inject_tcp_rst_after_recv_sub_ack.cpp

@@ -1,6 +1,6 @@
 #include <gtest/gtest.h>
 
-#include "packet_injector_test_frame.h"
+#include "packet_inject_test.h"
 
 TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_SUB_ACK)
 {
@@ -11,7 +11,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_SUB_ACK)
     snprintf(work_dir, sizeof(work_dir), "%s/%s", current_dir, "inject_ipv4_based_tcp_rst_after_recv_sub_ack");
     snprintf(input_dir, sizeof(input_dir), "%s/%s", current_dir, "pcap/inject_ipv4_based_tcp_rst_after_recv_sub_ack/test/");
 
-    struct packet_injector_case test = {
+    struct packet_inject_case test = {
         // descriptor
         .finish_clean_work_dir = 0,
         .descriptor = "Inject IPv4 based TCP RST after receiving SUB-ACK packet.",
@@ -44,7 +44,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_SUB_ACK)
         .diff_skip_pattern = "-I frame.time -I frame.time_epoch -I ip.id -I ip.ttl -I ip.checksum -I tcp.checksum -I tcp.window_size",
     };
 
-    packet_injector_test_frame_run(&test);
+    packet_inject_test(&test);
 }
 
 int main(int argc, char **argv)

test/packet_inject/gtest_inject_tcp_rst_after_recv_syn_ack.cpp

@@ -1,6 +1,6 @@
 #include <gtest/gtest.h>
 
-#include "packet_injector_test_frame.h"
+#include "packet_inject_test.h"
 
 TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_SYN_ACK)
 {
@@ -11,7 +11,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_SYN_ACK)
     snprintf(work_dir, sizeof(work_dir), "%s/%s", current_dir, "inject_ipv4_based_tcp_rst_after_recv_syn_ack");
     snprintf(input_dir, sizeof(input_dir), "%s/%s", current_dir, "pcap/inject_ipv4_based_tcp_rst_after_recv_syn_ack/test/");
 
-    struct packet_injector_case test = {
+    struct packet_inject_case test = {
         // descriptor
         .finish_clean_work_dir = 0,
         .descriptor = "Inject IPv4 based TCP RST after receiving SYN-ACK packet.",
@@ -44,7 +44,7 @@ TEST(INJECT_IPV4_BASED_TCP_RST, AFTER_RECV_SYN_ACK)
         .diff_skip_pattern = "-I frame.time -I frame.time_epoch -I ip.id -I ip.ttl -I ip.checksum -I tcp.checksum -I tcp.window_size",
     };
 
-    packet_injector_test_frame_run(&test);
+    packet_inject_test(&test);
 }
 
 int main(int argc, char **argv)

test/packet_inject/packet_inject_main.cpp

@@ -0,0 +1,166 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include "stellar_priv.h"
+#include "packet_inject_main.h"
+
+struct packet_inject_rule rule = {0};
+
+static void usage(char *cmd)
+{
+    printf("Usage: %s [options]\n\n", cmd);
+    printf("Options:\n");
+    printf("  -h <ip>       Host IP address\n");
+    printf("  -p <port>     Port number\n");
+    printf("  -t <type>     Type of manipulation\n");
+    printf("                   Options: tcp-rst, tcp-fin, tcp-payload, tcp-payload-fin-rst, udp-payload, ctrl-msg\n");
+    printf("  -c <condition> Condition for manipulation\n");
+    printf("                   Options: c2s-packet, s2c-packet\n");
+    printf("  -n <number>   Number of packets received before injecting action\n\n");
+    printf("Example:\n");
+    printf("  %s -h 192.168.1.100 -p 8080 -t tcp-payload -c c2s-packet -n 5\n", cmd);
+    printf("  %s -h 2001:db8::1   -p 8080 -t tcp-rst     -c s2c-packet -n 10\n", cmd);
+    printf("\n");
+}
+
+static int parse_cmd(int argc, char **argv)
+{
+    int opt = 0;
+    const char *host = NULL;
+    const char *type = NULL;
+    const char *condition = NULL;
+    while ((opt = getopt(argc, argv, "h:p:t:c:n:")) != -1)
+    {
+        switch (opt)
+        {
+        case 'h':
+            host = optarg;
+            break;
+        case 'p':
+            rule.port = htons(atoi(optarg));
+            break;
+        case 't':
+            type = optarg;
+            break;
+        case 'c':
+            condition = optarg;
+            break;
+        case 'n':
+            rule.number = atoi(optarg);
+            break;
+        default:
+            usage(argv[0]);
+            break;
+        }
+    }
+
+    if (host)
+    {
+        if (inet_pton(AF_INET, host, &rule.v4) != 1)
+        {
+            if (inet_pton(AF_INET6, host, &rule.v6) != 1)
+            {
+                printf("unable to convert host %s to IPv4 / IPv6\n", host);
+                return -1;
+            }
+            else
+            {
+                rule.ip_type = 6;
+            }
+        }
+        else
+        {
+            rule.ip_type = 4;
+        }
+    }
+
+    if (type == NULL)
+    {
+        usage(argv[0]);
+        printf("invalid type\n");
+        return -1;
+    }
+    else if (strcmp(type, "tcp-rst") == 0)
+    {
+        rule.inject_type = INJECT_TYPE_TCP_RST;
+    }
+    else if (strcmp(type, "tcp-fin") == 0)
+    {
+        rule.inject_type = INJECT_TYPE_TCP_FIN;
+    }
+    else if (strcmp(type, "tcp-payload") == 0)
+    {
+        rule.inject_type = INJECT_TYPE_TCP_PAYLOAD;
+    }
+    else if (strcmp(type, "tcp-payload-fin-rst") == 0)
+    {
+        rule.inject_type = INJECT_TYPE_TCP_PAYLOAD_FIN_RST;
+    }
+    else if (strcmp(type, "udp-payload") == 0)
+    {
+        rule.inject_type = INJECT_TYPE_UDP_PAYLOAD;
+    }
+    else if (strcmp(type, "ctrl-msg") == 0)
+    {
+        rule.inject_type = INJECT_TYPE_CTRL_MSG;
+    }
+    else
+    {
+        usage(argv[0]);
+        printf("invalid type\n");
+        return -1;
+    }
+
+    if (condition == NULL)
+    {
+        usage(argv[0]);
+        printf("invalid condition\n");
+        return -1;
+    }
+    else if (strcmp(condition, "c2s-packet") == 0)
+    {
+        rule.direction = AFTER_RECV_C2S_N_PACKET;
+    }
+    else if (strcmp(condition, "s2c-packet") == 0)
+    {
+        rule.direction = AFTER_RECV_S2C_N_PACKET;
+    }
+    else
+    {
+        usage(argv[0]);
+        printf("invalid condition\n");
+        return -1;
+    }
+
+    if (rule.number <= 0)
+    {
+        usage(argv[0]);
+        printf("invalid count\n");
+        return -1;
+    }
+    printf("%s load inject rule:\n", argv[0]);
+    printf("      host      : %s\n", host);
+    printf("      port      : %d\n", ntohs(rule.port));
+    printf("      type      : %s\n", type);
+    printf("      condition : %s\n", condition);
+    printf("      count     : %lu\n\n", rule.number);
+
+    return 0;
+}
+
+int packet_inject_main(int argc, char **argv)
+{
+    if (parse_cmd(argc, argv) != 0)
+    {
+        return -1;
+    }
+
+    return stellar_main(argc, argv);
+}
+
+int __attribute__((weak)) main(int argc, char **argv)
+{
+    return packet_inject_main(argc, argv);
+}

test/packet_inject/packet_inject_main.h

@@ -0,0 +1,46 @@
+#ifndef _PACKET_INJECT_MAIN_H
+#define _PACKET_INJECT_MAIN_H
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include <arpa/inet.h>
+
+#define AFTER_RECV_C2S_N_PACKET 1
+#define AFTER_RECV_S2C_N_PACKET 2
+
+enum packet_inject_type
+{
+    INJECT_TYPE_TCP_RST = 1,
+    INJECT_TYPE_TCP_FIN = 2,
+    INJECT_TYPE_TCP_PAYLOAD = 3,
+    INJECT_TYPE_TCP_PAYLOAD_FIN_RST = 4,
+    INJECT_TYPE_UDP_PAYLOAD = 5,
+    INJECT_TYPE_CTRL_MSG = 6,
+};
+
+struct packet_inject_rule
+{
+    int ip_type;
+    struct in_addr v4;  /* network order */
+    struct in6_addr v6; /* network order */
+    int port;           /* network order */
+
+    enum packet_inject_type inject_type;
+
+    // inject packet after (C2S/S2C) receiving n packets
+    int direction;   // AFTER_RECV_C2S_N_PACKET or AFTER_RECV_S2C_N_PACKET
+    uint64_t number; // n packets received
+};
+
+extern struct packet_inject_rule rule;
+
+int packet_inject_main(int argc, char **argv);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

test/packet_inject/packet_inject_plugin.cpp

@@ -0,0 +1,146 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "stellar/tuple.h"
+#include "stellar/session_mq.h"
+#include "packet_inject_main.h"
+
+struct packet_inject_plugin_ctx
+{
+    struct stellar *st;
+    int sess_plug_id;
+    int tcp_topic_id;
+    int udp_topic_id;
+    char name[64];
+};
+
+static void *on_sess_new(struct session *sess, void *plugin_ctx)
+{
+    struct packet_inject_plugin_ctx *ctx = (struct packet_inject_plugin_ctx *)plugin_ctx;
+    printf("[%s] pluign handle session new: %s\n", ctx->name, session_get0_readable_addr(sess));
+    return NULL;
+}
+
+static void on_sess_free(struct session *sess, void *sess_ctx, void *plugin_ctx)
+{
+    struct packet_inject_plugin_ctx *ctx = (struct packet_inject_plugin_ctx *)plugin_ctx;
+    printf("[%s] pluign handle session free: %s\n", ctx->name, session_get0_readable_addr(sess));
+}
+
+static void on_sess_msg(struct session *sess, int topic_id, const void *msg, void *sess_ctx, void *plugin_ctx)
+{
+    struct packet_inject_plugin_ctx *ctx = (struct packet_inject_plugin_ctx *)plugin_ctx;
+    printf("[%s] pluign handle session msg: %s (C2S received packets: %lu, S2C received packets: %lu)\n",
+           ctx->name, session_get0_readable_addr(sess),
+           session_get_stat(sess, FLOW_DIRECTION_C2S, STAT_RAW_PACKETS_RECEIVED),
+           session_get_stat(sess, FLOW_DIRECTION_S2C, STAT_RAW_PACKETS_RECEIVED));
+
+    char buffer[1024] = {0};
+    // struct packet *pkt = (struct packet *)msg;
+    const struct tuple6 *tuple = session_get_tuple6(sess);
+    if (rule.ip_type == 4 &&
+        memcmp(&tuple->src_addr.v4, &rule.v4, sizeof(struct in_addr)) &&
+        memcmp(&tuple->dst_addr.v4, &rule.v4, sizeof(struct in_addr)))
+    {
+        return;
+    }
+    if (rule.ip_type == 6 &&
+        memcmp(&tuple->src_addr.v6, &rule.v6, sizeof(struct in6_addr)) &&
+        memcmp(&tuple->dst_addr.v6, &rule.v6, sizeof(struct in6_addr)))
+    {
+        return;
+    }
+    if (rule.port != 0 && tuple->src_port != rule.port && tuple->dst_port != rule.port)
+    {
+        return;
+    }
+    if (session_get_stat(sess, FLOW_DIRECTION_C2S, STAT_INJECTED_PACKETS_SUCCESS) > 0 ||
+        session_get_stat(sess, FLOW_DIRECTION_S2C, STAT_INJECTED_PACKETS_SUCCESS) > 0)
+    {
+        return;
+    }
+    if (rule.direction == AFTER_RECV_C2S_N_PACKET && session_get_stat(sess, FLOW_DIRECTION_C2S, STAT_RAW_PACKETS_RECEIVED) != rule.number)
+    {
+        return;
+    }
+    if (rule.direction == AFTER_RECV_S2C_N_PACKET && session_get_stat(sess, FLOW_DIRECTION_S2C, STAT_RAW_PACKETS_RECEIVED) != rule.number)
+    {
+        return;
+    }
+    switch (rule.inject_type)
+    {
+    case INJECT_TYPE_TCP_RST:
+        stellar_inject_tcp_rst(ctx->st, sess, FLOW_DIRECTION_C2S);
+        stellar_inject_tcp_rst(ctx->st, sess, FLOW_DIRECTION_S2C);
+        session_set_discard(sess);
+        break;
+    case INJECT_TYPE_TCP_FIN:
+        stellar_inject_tcp_fin(ctx->st, sess, FLOW_DIRECTION_C2S);
+        stellar_inject_tcp_fin(ctx->st, sess, FLOW_DIRECTION_S2C);
+        session_set_discard(sess);
+        break;
+    case INJECT_TYPE_TCP_PAYLOAD:
+        snprintf(buffer, sizeof(buffer), "HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n%s", 5 + 5 + 2, "Hello");
+        stellar_inject_payload(ctx->st, sess, FLOW_DIRECTION_S2C, buffer, strlen(buffer)); // inject payload to client
+        stellar_inject_payload(ctx->st, sess, FLOW_DIRECTION_S2C, "World\r\n", 7);         // inject payload to client
+        stellar_inject_tcp_rst(ctx->st, sess, FLOW_DIRECTION_S2C);                         // inject RST to client
+        stellar_inject_tcp_rst(ctx->st, sess, FLOW_DIRECTION_C2S);                         // inject RST to server
+        session_set_discard(sess);
+        break;
+    case INJECT_TYPE_TCP_PAYLOAD_FIN_RST:
+        snprintf(buffer, sizeof(buffer), "HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n%s", 5 + 5 + 2, "Hello");
+        stellar_inject_payload(ctx->st, sess, FLOW_DIRECTION_S2C, buffer, strlen(buffer)); // inject payload to client
+        stellar_inject_payload(ctx->st, sess, FLOW_DIRECTION_S2C, "World\r\n", 7);         // inject payload to client
+        stellar_inject_tcp_fin(ctx->st, sess, FLOW_DIRECTION_S2C);                         // inject FIN to client
+        stellar_inject_tcp_rst(ctx->st, sess, FLOW_DIRECTION_S2C);                         // inject RST to client
+        stellar_inject_tcp_fin(ctx->st, sess, FLOW_DIRECTION_C2S);                         // inject FIN to server
+        stellar_inject_tcp_rst(ctx->st, sess, FLOW_DIRECTION_C2S);                         // inject RST to server
+        session_set_discard(sess);
+        break;
+    case INJECT_TYPE_UDP_PAYLOAD:
+        stellar_inject_payload(ctx->st, sess, FLOW_DIRECTION_C2S, "Hello Server", 12);
+        stellar_inject_payload(ctx->st, sess, FLOW_DIRECTION_S2C, "Hello Client", 12);
+        session_set_discard(sess);
+        break;
+    case INJECT_TYPE_CTRL_MSG:
+        // TOOD
+        break;
+    default:
+        break;
+    }
+}
+
+extern "C"
+{
+    void *packet_inject_plugin_init(struct stellar *st)
+    {
+        struct packet_inject_plugin_ctx *ctx = (struct packet_inject_plugin_ctx *)calloc(1, sizeof(struct packet_inject_plugin_ctx));
+        if (ctx == NULL)
+        {
+            return NULL;
+        }
+        ctx->st = st;
+        ctx->sess_plug_id = stellar_session_plugin_register(st, on_sess_new, on_sess_free, ctx);
+        ctx->tcp_topic_id = stellar_session_mq_get_topic_id(st, TOPIC_TCP);
+        ctx->udp_topic_id = stellar_session_mq_get_topic_id(st, TOPIC_UDP);
+        snprintf(ctx->name, sizeof(ctx->name), "packet_inject");
+
+        stellar_session_mq_subscribe(st, ctx->tcp_topic_id, on_sess_msg, ctx->sess_plug_id);
+        stellar_session_mq_subscribe(st, ctx->udp_topic_id, on_sess_msg, ctx->sess_plug_id);
+
+        printf("[%s] plugin init\n", ctx->name);
+
+        return ctx;
+    }
+
+    void packet_inject_plugin_exit(void *plugin_ctx)
+    {
+        struct packet_inject_plugin_ctx *ctx = (struct packet_inject_plugin_ctx *)plugin_ctx;
+        if (ctx)
+        {
+            printf("[%s] plugin exit\n", ctx->name);
+            free(ctx);
+        }
+    }
+}

test/packet_inject/packet_inject_test.cpp

@@ -1,9 +1,20 @@
-#include <gtest/gtest.h>
-
 #include <stdarg.h>
 #include <unistd.h>
 #include <sys/stat.h>
-#include "packet_injector_test_frame.h"
+#include <gtest/gtest.h>
+
+#include "packet_inject_test.h"
+#include "packet_inject_main.h"
+
+static int args_len(const char **args)
+{
+    int i = 0;
+    while (args[i] != NULL)
+    {
+        i++;
+    }
+    return i;
+}
 
 static void system_cmd(const char *cmd, ...)
 {
@@ -115,19 +126,8 @@ static void expect_cmp_inject(const char *expect_pcap_file, const char *inject_p
     stat(diff_json_txt, &s);
     EXPECT_TRUE(s.st_size == 0);
 }
-extern int packet_injector_main(int argc, char **argv);
 
-static int args_len(const char **args)
-{
-    int i = 0;
-    while (args[i] != NULL)
-    {
-        i++;
-    }
-    return i;
-}
-
-void packet_injector_test_frame_run(struct packet_injector_case *test)
+void packet_inject_test(struct packet_inject_case *test)
 {
     printf("\033[32m ============================================= \033[0m\n");
     printf("\033[32mTest: %s\033[0m\n", test->descriptor);
@@ -150,6 +150,8 @@ void packet_injector_test_frame_run(struct packet_injector_case *test)
     }
     system_cmd("cp %s/%s %s", test->input_prefix, test->input_pcap, dumpfile_dir);
     system_cmd("cp -r conf %s/", test->work_dir);
+    system_cmd("cp -r stellar_plugin %s/", test->work_dir);
+    system_cmd("cp -r libpacket_inject_plugin.so %s/", test->work_dir);
 
     // run packet injector
     char cwd[2048] = {0};
@@ -159,7 +161,7 @@ void packet_injector_test_frame_run(struct packet_injector_case *test)
     snprintf(temp, sizeof(temp), "dumpfile_dir = \"%s\"", dumpfile_dir);
     EXPECT_TRUE(replace_file_string("./conf/stellar.toml", "mode = marsio", "mode = dumpfile") == 0);
     EXPECT_TRUE(replace_file_string("./conf/stellar.toml", "dumpfile_dir = \"/tmp/dumpfile/\"", temp) == 0);
-    packet_injector_main(args_len(test->packet_injector_cmd), (char **)test->packet_injector_cmd);
+    packet_inject_main(args_len(test->packet_injector_cmd), (char **)test->packet_injector_cmd);
 
     // compare pcap
     for (int i = 0; i < MAX_COMPARISON; i++)

test/packet_inject/packet_inject_test.h

@@ -1,5 +1,5 @@
-#ifndef _PACKET_INJECTOR_TEST_FRAME_H
-#define _PACKET_INJECTOR_TEST_FRAME_H
+#ifndef _PACKET_INJECT_TEST_H
+#define _PACKET_INJECT_TEST_H
 
 #ifdef __cplusplus
 extern "C"
@@ -8,32 +8,26 @@ extern "C"
 
 #define MAX_COMPARISON 16
 
-struct packet_injector_case
+struct packet_inject_case
 {
-    // descriptor
     int finish_clean_work_dir;
     const char *descriptor;
     const char *work_dir;
 
-    // prefix
     const char *input_prefix;
-
-    // input pcap
     const char *input_pcap;
 
-    // compare
     struct
     {
         const char *expect_pcap;
         const char *inject_pcap;
     } compares[MAX_COMPARISON];
 
-    // packet injector command
     const char *packet_injector_cmd[16];
     const char *diff_skip_pattern;
 };
 
-void packet_injector_test_frame_run(struct packet_injector_case *test);
+void packet_inject_test(struct packet_inject_case *test);
 
 #ifdef __cplusplus
 }

test/packet_inject/stellar_plugin/spec.toml

@@ -0,0 +1,4 @@
+[[plugin]]
+path = "./libpacket_inject_plugin.so"
+init = "packet_inject_plugin_init"
+exit = "packet_inject_plugin_exit"

test/packet_inject/version.map

@@ -0,0 +1,7 @@
+LIBPACKET_INJECT_PLUGIN {
+global:
+	packet_inject_plugin_init;
+	packet_inject_plugin_exit;
+
+local: *;
+};

test/packet_injector.cpp

@@ -1,465 +0,0 @@
-#include <gtest/gtest.h>
-#include <stdio.h>
-#include <errno.h>
-#include <unistd.h>
-#include <string.h>
-#include <signal.h>
-#include <stdlib.h>
-#include <netinet/tcp.h>
-
-#include "logo.h"
-#include "times.h"
-#include "config.h"
-#include "id_generator.h"
-#include "stellar_priv.h"
-#include "session_priv.h"
-#include "inject_priv.h"
-#include "stellar/tuple.h"
-#include "stellar/session_mq.h"
-
-/******************************************************************************
- * packet_injector
- ******************************************************************************/
-
-enum condition
-{
-    AFTER_RECV_C2S_N_PACKET = 1,
-    AFTER_RECV_S2C_N_PACKET = 2,
-};
-
-enum inject_type
-{
-    INJECT_TYPE_TCP_RST = 1,
-    INJECT_TYPE_TCP_FIN = 2,
-    INJECT_TYPE_TCP_PAYLOAD = 3,
-    INJECT_TYPE_TCP_PAYLOAD_FIN_RST = 4,
-    INJECT_TYPE_UDP_PAYLOAD = 5,
-    INJECT_TYPE_CTRL_MSG = 6,
-};
-
-struct inject_rule
-{
-    int ip_type;
-    struct in_addr v4;  /* network order */
-    struct in6_addr v6; /* network order */
-    int port;           /* network order */
-
-    enum inject_type inject_type;
-    enum condition count_dir;
-    uint64_t count_num;
-} rule;
-
-#define INJECT_PACKET_PLUGIN_LOG_DEBUG(format, ...) LOG_DEBUG("inject packet plugin", format, ##__VA_ARGS__)
-
-static void usage(char *cmd)
-{
-    printf("Usage: %s [options]\n\n", cmd);
-    printf("Options:\n");
-    printf("  -h <ip>       Host IP address\n");
-    printf("  -p <port>     Port number\n");
-    printf("  -t <type>     Type of manipulation\n");
-    printf("                   Options: tcp-rst, tcp-fin, tcp-payload, tcp-payload-fin-rst, udp-payload, ctrl-msg\n");
-    printf("  -c <condition> Condition for manipulation\n");
-    printf("                   Options: c2s-packet, s2c-packet\n");
-    printf("  -n <number>   Number of packets received before injecting action\n\n");
-    printf("Example:\n");
-    printf("  %s -h 192.168.1.100 -p 8080 -t tcp-payload -c c2s-packet -n 5\n", cmd);
-    printf("  %s -h 2001:db8::1   -p 8080 -t tcp-rst     -c s2c-packet -n 10\n", cmd);
-    printf("\n");
-}
-
-static int packet_injector_on_init(int argc, char **argv, struct inject_rule *rule)
-{
-    memset(rule, 0, sizeof(struct inject_rule));
-
-    int opt = 0;
-    const char *host = NULL;
-    const char *type = NULL;
-    const char *condition = NULL;
-    while ((opt = getopt(argc, argv, "h:p:t:c:n:")) != -1)
-    {
-        switch (opt)
-        {
-        case 'h':
-            host = optarg;
-            break;
-        case 'p':
-            rule->port = htons(atoi(optarg));
-            break;
-        case 't':
-            type = optarg;
-            break;
-        case 'c':
-            condition = optarg;
-            break;
-        case 'n':
-            rule->count_num = atoi(optarg);
-            break;
-        default:
-            usage(argv[0]);
-            break;
-        }
-    }
-
-    if (host)
-    {
-        if (inet_pton(AF_INET, host, &rule->v4) != 1)
-        {
-            if (inet_pton(AF_INET6, host, &rule->v6) != 1)
-            {
-                printf("unable to convert host %s to IPv4 / IPv6\n", host);
-                return -1;
-            }
-            else
-            {
-                rule->ip_type = 6;
-            }
-        }
-        else
-        {
-            rule->ip_type = 4;
-        }
-    }
-
-    if (type == NULL)
-    {
-        usage(argv[0]);
-        printf("invalid type\n");
-        return -1;
-    }
-    else if (strcmp(type, "tcp-rst") == 0)
-    {
-        rule->inject_type = INJECT_TYPE_TCP_RST;
-    }
-    else if (strcmp(type, "tcp-fin") == 0)
-    {
-        rule->inject_type = INJECT_TYPE_TCP_FIN;
-    }
-    else if (strcmp(type, "tcp-payload") == 0)
-    {
-        rule->inject_type = INJECT_TYPE_TCP_PAYLOAD;
-    }
-    else if (strcmp(type, "tcp-payload-fin-rst") == 0)
-    {
-        rule->inject_type = INJECT_TYPE_TCP_PAYLOAD_FIN_RST;
-    }
-    else if (strcmp(type, "udp-payload") == 0)
-    {
-        rule->inject_type = INJECT_TYPE_UDP_PAYLOAD;
-    }
-    else if (strcmp(type, "ctrl-msg") == 0)
-    {
-        rule->inject_type = INJECT_TYPE_CTRL_MSG;
-    }
-    else
-    {
-        usage(argv[0]);
-        printf("invalid type\n");
-        return -1;
-    }
-
-    if (condition == NULL)
-    {
-        usage(argv[0]);
-        printf("invalid condition\n");
-        return -1;
-    }
-    else if (strcmp(condition, "c2s-packet") == 0)
-    {
-        rule->count_dir = AFTER_RECV_C2S_N_PACKET;
-    }
-    else if (strcmp(condition, "s2c-packet") == 0)
-    {
-        rule->count_dir = AFTER_RECV_S2C_N_PACKET;
-    }
-    else
-    {
-        usage(argv[0]);
-        printf("invalid condition\n");
-        return -1;
-    }
-
-    if (rule->count_num <= 0)
-    {
-        usage(argv[0]);
-        printf("invalid count\n");
-        return -1;
-    }
-    printf("%s load inject rule:\n", argv[0]);
-    printf("      host      : %s\n", host);
-    printf("      port      : %d\n", ntohs(rule->port));
-    printf("      type      : %s\n", type);
-    printf("      condition : %s\n", condition);
-    printf("      count     : %lu\n\n", rule->count_num);
-
-    return 0;
-}
-
-static void packet_injector_on_msg(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env)
-{
-    char buffer[1024] = {0};
-    struct inject_rule *p_rule = &rule;
-    // struct packet *pkt = (struct packet *)msg;
-    const struct tuple6 *tuple = session_get_tuple6(sess);
-    if (p_rule->ip_type == 4 &&
-        memcmp(&tuple->src_addr.v4, &p_rule->v4, sizeof(struct in_addr)) &&
-        memcmp(&tuple->dst_addr.v4, &p_rule->v4, sizeof(struct in_addr)))
-    {
-        return;
-    }
-    if (p_rule->ip_type == 6 &&
-        memcmp(&tuple->src_addr.v6, &p_rule->v6, sizeof(struct in6_addr)) &&
-        memcmp(&tuple->dst_addr.v6, &p_rule->v6, sizeof(struct in6_addr)))
-    {
-        return;
-    }
-    if (p_rule->port != 0 && tuple->src_port != p_rule->port && tuple->dst_port != p_rule->port)
-    {
-        return;
-    }
-    if (session_get_stat(sess, FLOW_DIRECTION_C2S, STAT_INJECTED_PACKETS_SUCCESS) > 0 ||
-        session_get_stat(sess, FLOW_DIRECTION_S2C, STAT_INJECTED_PACKETS_SUCCESS) > 0)
-    {
-        return;
-    }
-    if (p_rule->count_dir == AFTER_RECV_C2S_N_PACKET && session_get_stat(sess, FLOW_DIRECTION_C2S, STAT_RAW_PACKETS_RECEIVED) != p_rule->count_num)
-    {
-        return;
-    }
-    if (p_rule->count_dir == AFTER_RECV_S2C_N_PACKET && session_get_stat(sess, FLOW_DIRECTION_S2C, STAT_RAW_PACKETS_RECEIVED) != p_rule->count_num)
-    {
-        return;
-    }
-    switch (p_rule->inject_type)
-    {
-    case INJECT_TYPE_TCP_RST:
-        EXPECT_TRUE(stellar_inject_tcp_rst(sess, FLOW_DIRECTION_C2S) > 0);
-        EXPECT_TRUE(stellar_inject_tcp_rst(sess, FLOW_DIRECTION_S2C) > 0);
-        session_set_discard(sess);
-        break;
-    case INJECT_TYPE_TCP_FIN:
-        EXPECT_TRUE(stellar_inject_tcp_fin(sess, FLOW_DIRECTION_C2S) > 0);
-        EXPECT_TRUE(stellar_inject_tcp_fin(sess, FLOW_DIRECTION_S2C) > 0);
-        session_set_discard(sess);
-        break;
-    case INJECT_TYPE_TCP_PAYLOAD:
-        snprintf(buffer, sizeof(buffer), "HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n%s", 5 + 5 + 2, "Hello");
-        EXPECT_TRUE(stellar_inject_payload(sess, FLOW_DIRECTION_S2C, buffer, strlen(buffer)) > 0); // inject payload to client
-        EXPECT_TRUE(stellar_inject_payload(sess, FLOW_DIRECTION_S2C, "World\r\n", 7) > 0);         // inject payload to client
-        EXPECT_TRUE(stellar_inject_tcp_rst(sess, FLOW_DIRECTION_S2C) > 0);                         // inject RST to client
-        EXPECT_TRUE(stellar_inject_tcp_rst(sess, FLOW_DIRECTION_C2S) > 0);                         // inject RST to server
-        session_set_discard(sess);
-        break;
-    case INJECT_TYPE_TCP_PAYLOAD_FIN_RST:
-        snprintf(buffer, sizeof(buffer), "HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n%s", 5 + 5 + 2, "Hello");
-        EXPECT_TRUE(stellar_inject_payload(sess, FLOW_DIRECTION_S2C, buffer, strlen(buffer)) > 0); // inject payload to client
-        EXPECT_TRUE(stellar_inject_payload(sess, FLOW_DIRECTION_S2C, "World\r\n", 7) > 0);         // inject payload to client
-        EXPECT_TRUE(stellar_inject_tcp_fin(sess, FLOW_DIRECTION_S2C) > 0);                         // inject FIN to client
-        EXPECT_TRUE(stellar_inject_tcp_rst(sess, FLOW_DIRECTION_S2C) > 0);                         // inject RST to client
-        EXPECT_TRUE(stellar_inject_tcp_fin(sess, FLOW_DIRECTION_C2S) > 0);                         // inject FIN to server
-        EXPECT_TRUE(stellar_inject_tcp_rst(sess, FLOW_DIRECTION_C2S) > 0);                         // inject RST to server
-        session_set_discard(sess);
-        break;
-    case INJECT_TYPE_UDP_PAYLOAD:
-        EXPECT_TRUE(stellar_inject_payload(sess, FLOW_DIRECTION_C2S, "Hello Server", 12) > 0);
-        EXPECT_TRUE(stellar_inject_payload(sess, FLOW_DIRECTION_S2C, "Hello Client", 12) > 0);
-        session_set_discard(sess);
-        break;
-    case INJECT_TYPE_CTRL_MSG:
-        // TOOD
-        break;
-    default:
-        break;
-    }
-}
-
-static void *packet_injector_on_sess_new(struct session *sess, void *plugin_env)
-{
-    return NULL;
-}
-
-static void packet_injector_on_sess_free(struct session *sess, void *ctx, void *plugin_env)
-{
-    char buff[4096] = {0};
-    session_to_json(sess, buff, sizeof(buff));
-    INJECT_PACKET_PLUGIN_LOG_DEBUG("=> session: %s", buff);
-}
-
-/******************************************************************************
- * main
- ******************************************************************************/
-
-static const char *log_config_file = "./conf/log.toml";
-static const char *stellar_config_file = "./conf/stellar.toml";
-
-static void signal_handler(int signo)
-{
-    if (signo == SIGINT)
-    {
-        STELLAR_LOG_STATE("SIGINT received, notify threads to exit !!!");
-        ATOMIC_SET(&runtime->need_exit, 1);
-    }
-
-    if (signo == SIGQUIT)
-    {
-        STELLAR_LOG_STATE("SIGQUIT received, notify threads to exit !!!");
-        ATOMIC_SET(&runtime->need_exit, 1);
-    }
-
-    if (signo == SIGTERM)
-    {
-        STELLAR_LOG_STATE("SIGTERM received, notify threads to exit !!!");
-        ATOMIC_SET(&runtime->need_exit, 1);
-    }
-
-    if (signo == SIGHUP)
-    {
-        STELLAR_LOG_STATE("SIGHUP received, reload log level !!!");
-        log_reload_level(log_config_file);
-    }
-}
-
-static int all_session_have_freed(void)
-{
-    for (int i = 0; i < config->io_opts.nr_threads; i++)
-    {
-        struct session_manager *sess_mgr = runtime->threads[i].sess_mgr;
-        struct session_manager_stat *sess_stat = session_manager_stat(sess_mgr);
-
-        if (ATOMIC_READ(&sess_stat->curr_nr_tcp_sess_used) != 0)
-        {
-            return 0;
-        }
-
-        if (ATOMIC_READ(&sess_stat->curr_nr_udp_sess_used) != 0)
-        {
-            return 0;
-        }
-    }
-
-    return 1;
-}
-
-static int all_stat_have_output(void)
-{
-    static int count = 0;
-    if (runtime->stat_last_output_ts == stellar_get_monotonic_time_msec())
-    {
-        count++;
-    }
-
-    return count == 2;
-}
-
-int packet_injector_main(int argc, char **argv)
-{
-    if (packet_injector_on_init(argc, argv, &rule) != 0)
-    {
-        return -1;
-    }
-
-    stellar_update_time_cache();
-    struct stellar st = {runtime};
-    int sess_plug_id;
-    int tcp_topic_id;
-    int udp_topic_id;
-
-    signal(SIGINT, signal_handler);
-    signal(SIGQUIT, signal_handler);
-    signal(SIGTERM, signal_handler);
-    signal(SIGHUP, signal_handler);
-
-    if (log_init(log_config_file) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to init log");
-        goto error_out;
-    }
-
-    if (stellar_load_config(stellar_config_file, config) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to load config file");
-        goto error_out;
-    }
-    stellar_print_config(config);
-    STELLAR_LOG_DEBUG("sizeof(struct session) = %lu bytes", sizeof(struct session));
-
-    if (id_generator_init(config->dev_opts.base, config->dev_opts.offset) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to init id generator");
-        goto error_out;
-    }
-
-    runtime->stat = stellar_stat_new(config->io_opts.nr_threads);
-    if (runtime->stat == NULL)
-    {
-        STELLAR_LOG_ERROR("unable to create stellar stat");
-        goto error_out;
-    }
-    runtime->plug_mgr = plugin_manager_init(&st, "./stellar_plugin/spec.toml");
-    if (runtime->plug_mgr == NULL)
-    {
-        STELLAR_LOG_ERROR("unable to create plugin manager");
-        goto error_out;
-    }
-
-    runtime->packet_io = packet_io_new(&config->io_opts);
-    if (runtime->packet_io == NULL)
-    {
-        STELLAR_LOG_ERROR("unable to create packet io");
-        goto error_out;
-    }
-
-    sess_plug_id = stellar_session_plugin_register(&st, packet_injector_on_sess_new, packet_injector_on_sess_free, NULL);
-    tcp_topic_id = stellar_session_mq_get_topic_id(&st, TOPIC_TCP);
-    udp_topic_id = stellar_session_mq_get_topic_id(&st, TOPIC_UDP);
-
-    stellar_session_mq_subscribe(&st, tcp_topic_id, packet_injector_on_msg, sess_plug_id);
-    stellar_session_mq_subscribe(&st, udp_topic_id, packet_injector_on_msg, sess_plug_id);
-
-    if (stellar_thread_init(runtime, config) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to init thread context");
-        goto error_out;
-    }
-
-    if (stellar_thread_run(runtime, config) != 0)
-    {
-        STELLAR_LOG_ERROR("unable to create worker thread");
-        goto error_out;
-    }
-
-    runtime->stat_last_output_ts = stellar_get_monotonic_time_msec();
-    while (!ATOMIC_READ(&runtime->need_exit))
-    {
-        stellar_update_time_cache();
-        if (stellar_get_monotonic_time_msec() - runtime->stat_last_output_ts > 2000)
-        {
-            runtime->stat_last_output_ts = stellar_get_monotonic_time_msec();
-            stellar_stat_output(runtime->stat);
-        }
-        usleep(1000); // 1ms
-
-        // Only available in dump file mode, automatically exits when all sessions have been released
-        if (packet_io_wait_exit(runtime->packet_io) && all_session_have_freed() && all_stat_have_output())
-        {
-            STELLAR_LOG_STATE("all sessions have been released and all stat have been output, notify threads to exit !!!");
-            ATOMIC_SET(&runtime->need_exit, 1);
-        }
-    }
-
-error_out:
-    stellar_thread_join(runtime, config);
-    stellar_thread_clean(runtime, config);
-    packet_io_free(runtime->packet_io);
-    plugin_manager_exit(runtime->plug_mgr);
-    stellar_stat_free(runtime->stat);
-    STELLAR_LOG_STATE("stellar exit !!!\n");
-    log_free();
-
-    return 0;
-}
-
-int __attribute__((weak)) main(int argc, char **argv)
-{
-    return packet_injector_main(argc, argv);
-}

test/packet_parser/CMakeLists.txt

@@ -0,0 +1,3 @@
+# build packet_parser
+add_executable(packet_parser packet_parser.cpp)
+target_link_libraries(packet_parser packet pcap)