gfwleak/niubinghui-luapluginmanage
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【修改】删除就版本中部分失效文件

Browse Source
This commit is contained in:
niubinghui
2024-10-15 19:15:38 +08:00
parent 7c8146f00f
commit 9c39d69b21
17 changed files with 0 additions and 1384 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
config/lua_plugin_manage.toml
View File

@@ -1,10 +0,0 @@
# config.toml
[[plugin]]
path = "example_plugin-1.lua"
init = "plugin_load"
exit = "plugin_unload"
[[plugin]]
path = "example_plugin-2.lua"
init = "plugin_load"
exit = "plugin_unload"

18
include/lua_plugin_manage.h
View File

@@ -1,18 +0,0 @@
#pragma once
#include "stellar/stellar.h"
#include <stddef.h>
struct lua_config_spec
{
char *spec_file_path; /* 插件需要使用的文件名 */
char *spec_load_func_name; /* 加载插件需要调用的函数名称 */
char *spec_unload_func_name; /* 卸载插件需要调用的函数名称 */
};
struct lua_plugin_manage;
struct lua_plugin_manage *lua_plugin_manage_init(struct stellar *st, struct lua_config_spec specifics[], size_t specific_num);
void lua_plugin_manage_exit(struct lua_plugin_manage *lua_plug_mgr);
// void debug_lua_plugin_manage_schema(struct lua_plugin_manage *schema);

36
test/Makefile
View File

@@ -1,36 +0,0 @@
TOPDIR = ./..
CC=gcc
MAKE=make
TARGET=example
SIMPLE=simple_stellar_plugin.so
EXAMPLE_FLAG = -DLUAPLUGIN_EXAMPLE
SRC := example_plugin_manage.c
OBJECTS := example_plugin_manage.o
SIMPLE_SRC := simple_example_plugin.c
SIMPLE_OBJECTS := simple_example_plugin.o
INCLUDE = -I$(TOPDIR)/output/include -I$(TOPDIR)/dependence/include -I$(TOPDIR)/test/include
CFLAGS = -g -Wextra -Wall -O0 -fPIC
# CFLAGS += -pedantic -fsanitize=address
# LDLIBS = -L$(TOPDIR)/output/lib -llua -ldl -lm
LDLIBS += -L$(TOPDIR)/output/libs -lluaplugin -L$(TOPDIR)/dependence/lib -ltoml -lbitmap -lplugin_manager -lstellar_on_sapp
TARGET:$(OBJECTS)
$(CC) $(CFLAGS) -o $(TARGET) $(OBJECTS) $(LDLIBS)
$(OBJECTS):$(SRC)
$(CC) $(TEST_FLAG) $(INCLUDE) $(CFLAGS) $(SRC) -c $^
SIMPLE:$(SIMPLE_OBJECTS)
$(CC) $(CFLAGS) --shared -o $(SIMPLE) $(SIMPLE_OBJECTS) $(LDLIBS)
cp $(SIMPLE) plugin
$(SIMPLE_OBJECTS):$(SIMPLE_SRC)
$(CC) $(TEST_FLAG) $(INCLUDE) $(CFLAGS) $(SIMPLE_SRC) -c $^
clean:
rm -rf $(OBJECTS) $(TARGET) $(SIMPLE_OBJECTS) $(SIMPLE)
rm -rf $(TOPDIR)/output/libs/$(TARGET)

15
test/conf/lua_plugin_manage.toml
View File

@@ -1,15 +0,0 @@
# config.toml
[[plugin]]
path = "./plugin/example_plugin_regist.lua"
init = "plugin_load"
exit = "plugin_unload"
[[plugin]]
path = "./plugin/example_plugin_topic.lua"
init = "plugin_load"
exit = "plugin_unload"
[[plugin]]
path = "./plugin/example_plugin_message.lua"
init = "plugin_load"
exit = "plugin_unload"

9
test/conf/plugin_manage.toml
View File

@@ -1,9 +0,0 @@
[[plugin]]
path = "./plugin/simple_stellar_plugin.so"
init = "simple_plugin_sub_session_stat_init"
exit = "simple_plugin_sub_session_stat_exit"
[[plugin]]
path = "./plugin/simple_stellar_plugin.so"
init = "simple_session_packet_plugin_init"
exit = "simple_session_packet_plugin_exit"

103
test/example.c
View File

@@ -1,103 +0,0 @@
#include "lua_plugin_manage.h"
#include <toml.h>
#include <utarray.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#define CONFIG_PATH "../output/conf/lua_plugin_manage.toml"
struct lua_config_specific * config_load(const char *config_file_name, int * specific_num);
int main()
{
struct stellar *st = stellar_new();
int num = 0;
struct lua_config_specific * specific = config_load(CONFIG_PATH, &num);
struct lua_plugin_manage_schema *schema = lua_plugin_manage_init(st, num, specific);
for (int i = 0; i < 1; ++i)
{
struct session *sess1 = session_new(10000, 20000);
struct registered_session_plugin_schema *plugin = NULL;
while ((plugin = utarray_next(st->plugin_array, plugin)))
{
printf("call plugin id %d\n", plugin->plugin_id);
sess1->plugin_id = plugin->plugin_id;
void *temp_pointer = plugin->on_ctx_new(sess1, plugin->plugin_env);
struct session_data_pair pair = {plugin->plugin_id, temp_pointer};
utarray_push_back(sess1->session_plugin, &pair);
printf("debug session: %d, %d\n", sess1->session_id, sess1->session_type);
}
plugin = NULL;
while ((plugin = utarray_next(st->plugin_array, plugin)))
{
printf("call plugin id %d\n", plugin->plugin_id);
sess1->plugin_id = plugin->plugin_id;
void *temp_context = session_get_private(sess1, plugin->plugin_id);
plugin->on_ctx_free(sess1, temp_context, plugin->plugin_env);
}
utarray_free(sess1->session_plugin);
free(sess1);
}
lua_plugin_manage_exit(schema);
return 0;
}
struct lua_config_specific * config_load(const char *config_file_name, int * specific_count)
{
if (__glibc_unlikely(!config_file_name))
return NULL;
int specific_num = 0;
char errbuff[256] = {0};
if (access(config_file_name, F_OK))
return NULL;
FILE *fp = fopen(config_file_name, "r");
if (!fp)
return NULL;
toml_table_t *conf = toml_parse_file(fp, errbuff, sizeof(errbuff));
if (fp)
fclose(fp);
if (!conf)
{
printf("parse config file failed, filename %s, err %s\n", config_file_name, errbuff);
return NULL;
}
toml_array_t *plugin_array = toml_array_in(conf, "plugin");
if (!plugin_array)
return NULL;
specific_num = toml_array_nelem(plugin_array);
struct lua_config_specific * new_spec = (struct lua_config_specific *)calloc(specific_num, sizeof(struct lua_config_specific));
if (!new_spec)
return NULL;
struct lua_config_specific * specific = NULL;
for (int i = 0; i < specific_num; ++i)
{
toml_table_t *plugin = toml_table_at(plugin_array, i);
const char *raw_filepath = toml_raw_in(plugin, "path");
const char *raw_load_func_name = toml_raw_in(plugin, "init");
const char *raw_unload_func_name = toml_raw_in(plugin, "exit");
specific = &new_spec[i];
if (toml_rtos(raw_filepath, &specific->config_specific_file) ||
toml_rtos(raw_load_func_name, &specific->config_specific_load_func) ||
toml_rtos(raw_unload_func_name, &specific->config_specific_unload_func))
{
toml_free(conf);
free(specific);
return NULL;
}
}
*specific_count = specific_num;
return new_spec;
}

286
test/example_plugin_manage.c
View File

@@ -1,286 +0,0 @@
#include "plugin_manager_gtest_mock.h"
#include "lua_plugin_manage.h"
#include <toml.h>
#include <utarray.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <assert.h>
#include <getopt.h>
#include <time.h>
#define PLUGIN_CONFIG_PATH "./conf/plugin_manage.toml"
#define LUA_CONFIG_PATH "./conf/lua_plugin_manage.toml"
// #define DEBUG_PLUGIN_SCHEMA
#define PACKET_COUNT 3
#define SESSION_COUNT 10
struct arg_config
{
int session_count;
int packet_count;
int wait_time;
int thread_num;
unsigned char lua_mod;
unsigned char session_mod;
unsigned char print_mod;
unsigned char time_mod;
char *config_path;
};
struct arg_config global_arg_config = {SESSION_COUNT, PACKET_COUNT, 10, 1, 1, 1, 1, 1, LUA_CONFIG_PATH};
static struct lua_config_spec *config_load(const char *config_file_name, int *specific_num);
static void parse_args(int argc, char *argv[]);
// #ifdef DEBUG_PLUGIN_SCHEMA
static void debug_plugin_manage_specific(struct lua_config_spec *specific, int num);
static void debug_plugin_manage_schema(struct plugin_manager_schema *schema);
// #endif
int main(int argc, char *argv[])
{
parse_args(argc, argv);
if (global_arg_config.print_mod)
{
printf("***** ***** ***** ***** ***** *****\n");
printf("load example config:\n");
printf("session count is %d\n", global_arg_config.session_count);
printf("packet count is %d\n", global_arg_config.packet_count);
printf("thread num is %d\n", global_arg_config.thread_num);
printf("load config from path %s\n", global_arg_config.config_path);
printf("***** ***** ***** ***** ***** *****\n\n");
}
clock_t start_clock, end_clock;
struct stellar st;
memset(&st, 0, sizeof(st));
st.thread_num = global_arg_config.thread_num;
/* ***** ***** 进行初始化 ***** ***** */
struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, PLUGIN_CONFIG_PATH);
assert(plug_mgr);
/* 初始化lua插件 */
// printf("now wait time to compute memory ... ...\n");
// sleep(global_arg_config.wait_time);
struct lua_plugin_manage *lua_schema = NULL;
if (global_arg_config.lua_mod)
{
int num = 0;
struct lua_config_spec *specific = config_load(LUA_CONFIG_PATH, &num);
assert(specific);
if (global_arg_config.print_mod)
debug_plugin_manage_specific(specific, num);
lua_schema = lua_plugin_manage_init(&st, specific, num);
st.lua_plug_mgr = lua_schema;
assert(lua_schema);
if (specific)
free(specific);
// printf("now wait time to compute memory ... ...\n");
// sleep(global_arg_config.wait_time);
}
if (global_arg_config.print_mod)
debug_plugin_manage_schema(plug_mgr);
/* ***** ***** 初始化完成 ***** ***** */
/* ***** ***** 会话相关测试 ***** ***** */
if (global_arg_config.session_mod)
{
unsigned char ip_proto = 6;
struct packet pkt = {&st, TCP, ip_proto};
struct session *sess = (struct session *)calloc(global_arg_config.session_count, sizeof(struct session));
memset(sess, 0, global_arg_config.session_count * sizeof(struct session));
/* 创建session */
for (int i = 0; i < global_arg_config.session_count; i++)
{
sess[i].plug_mgr_rt = plugin_manager_session_runtime_new(plug_mgr, &sess[i]);
sess[i].type = SESSION_TYPE_TCP;
}
clock_t total_clock = 0;
for (int j = 0; j < global_arg_config.packet_count; j++)
{
plugin_manager_on_packet_ingress(plug_mgr, &pkt);
for (int i = 0; i < global_arg_config.session_count; i++)
{
sess[i].sess_pkt_cnt += 1;
if (global_arg_config.time_mod)
start_clock = clock();
plugin_manager_on_session_ingress(&sess[i], &pkt);
plugin_manager_on_session_egress(&sess[i], &pkt);
if (global_arg_config.time_mod)
{
end_clock = clock();
total_clock += (end_clock - start_clock);
}
}
if (global_arg_config.time_mod)
{
printf("for session count %d, total time is %ld\n", global_arg_config.session_count, total_clock);
printf("trans to second, time is %f\n", ((double)total_clock / CLOCKS_PER_SEC));
}
plugin_manager_on_packet_egress(plug_mgr, &pkt);
}
/* 释放session */
for (int i = 0; i < global_arg_config.session_count; i++)
{
plugin_manager_on_session_closing(&sess[i]);
plugin_manager_session_runtime_free(sess[i].plug_mgr_rt);
}
}
/* ***** ***** 会话相关完成 ***** ***** */
/* ***** ***** 释放 ***** ***** */
lua_plugin_manage_exit(lua_schema);
plugin_manager_exit(plug_mgr);
/* ***** ***** 释放 ***** ***** */
return 0;
}
static struct lua_config_spec *config_load(const char *config_file_name, int *specific_count)
{
if (__glibc_unlikely(!config_file_name))
return NULL;
int specific_num = 0;
char errbuff[256] = {0};
if (access(config_file_name, F_OK))
return NULL;
FILE *fp = fopen(config_file_name, "r");
if (!fp)
return NULL;
toml_table_t *conf = toml_parse_file(fp, errbuff, sizeof(errbuff));
if (fp)
fclose(fp);
if (!conf)
{
printf("parse config file failed, filename %s, err %s\n", config_file_name, errbuff);
return NULL;
}
toml_array_t *plugin_array = toml_array_in(conf, "plugin");
if (!plugin_array)
return NULL;
specific_num = toml_array_nelem(plugin_array);
struct lua_config_spec *new_spec = (struct lua_config_spec *)calloc(specific_num, sizeof(struct lua_config_spec));
if (!new_spec)
return NULL;
struct lua_config_spec *specific = NULL;
for (int i = 0; i < specific_num; ++i)
{
toml_table_t *plugin = toml_table_at(plugin_array, i);
const char *raw_filepath = toml_raw_in(plugin, "path");
const char *raw_load_func_name = toml_raw_in(plugin, "init");
const char *raw_unload_func_name = toml_raw_in(plugin, "exit");
specific = &new_spec[i];
if (toml_rtos(raw_filepath, &specific->spec_file_path) ||
toml_rtos(raw_load_func_name, &specific->spec_load_func_name) ||
toml_rtos(raw_unload_func_name, &specific->spec_unload_func_name))
{
toml_free(conf);
free(specific);
return NULL;
}
}
*specific_count = specific_num;
return new_spec;
}
static void parse_args(int argc, char *argv[])
{
int opt;
static const char *shortopts = "s:p:w:c:t:lnde";
static struct option longopts[] = {
{"session_count", required_argument, NULL, 's'},
{"packet_count", required_argument, NULL, 'p'},
{"wait_time", required_argument, NULL, 'w'},
{"config_path", required_argument, NULL, 'c'},
{"thread_num", required_argument, NULL, 't'},
{"disable_lua", no_argument, NULL, 'l'},
{"disable_session", no_argument, NULL, 'n'},
{"print_debug", no_argument, NULL, 'd'},
{"disable_time", no_argument, NULL, 'e'},
{0, 0, 0, 0},
};
while ((opt = getopt_long(argc, argv, shortopts, longopts, NULL)) != -1)
{
switch (opt)
{
case 's':
global_arg_config.session_count = atoi(optarg);
break;
case 'p':
global_arg_config.packet_count = atoi(optarg);
break;
case 'w':
global_arg_config.wait_time = atoi(optarg);
break;
case 'c':
global_arg_config.config_path = strdup(optarg);
break;
case 't':
global_arg_config.thread_num = atoi(optarg);
break;
case 'l':
global_arg_config.lua_mod = 0;
break;
case 'n':
global_arg_config.session_mod = 0;
break;
case 'd':
global_arg_config.print_mod = 1;
break;
case 'e':
global_arg_config.time_mod = 0;
break;
default:
break;
}
}
}
// #ifdef DEBUG_PLUGIN_SCHEMA
static void debug_plugin_manage_specific(struct lua_config_spec *specific, int num)
{
printf("***** ***** ***** ***** ***** *****\n");
printf("load config specifics count is %d\n", num);
for (int i = 0; i < num; ++i)
{
printf("[%d]file path: %s\n", i, specific[i].spec_file_path);
printf("[%d]load func: %s\n", i, specific[i].spec_load_func_name);
printf("[%d]unload func: %s\n", i, specific[i].spec_unload_func_name);
}
printf("***** ***** ***** ***** ***** *****\n\n");
}
static void debug_plugin_manage_schema(struct plugin_manager_schema *schema)
{
struct registered_session_plugin_schema *plugin = NULL;
for (int i = 0; i < (int)utarray_len(schema->registered_session_plugin_array); ++i)
{
plugin = (struct registered_session_plugin_schema *)utarray_eltptr(schema->registered_session_plugin_array, (unsigned int)i);
printf("plugin[%d]: new func %p, free func %p, env %p\n", i, plugin->on_ctx_new, plugin->on_ctx_free, plugin->plugin_env);
}
printf("\n");
return;
}
// #endif

22
test/include/plugin_manager.h
View File

@@ -1,22 +0,0 @@
#pragma once
#include "stellar/stellar.h"
struct plugin_manager_schema;
struct plugin_manager_runtime;
struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char *plugin_spec_file_path);
void plugin_manager_exit(struct plugin_manager_schema *plug_mgr);
void plugin_manager_on_packet_ingress(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
void plugin_manager_on_packet_egress(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
//return polling work state, 0: idle, 1: working
int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr);
//publish and dispatch session msg(msg, pkt) on session_mq
void plugin_manager_on_session_ingress(struct session *sess,struct packet *pkt);
void plugin_manager_on_session_egress(struct session *sess,struct packet *pkt);
void plugin_manager_on_session_closing(struct session *sess);
struct plugin_manager_runtime *plugin_manager_session_runtime_new(struct plugin_manager_schema *plug_mgr, struct session *sess);
void plugin_manager_session_runtime_free(struct plugin_manager_runtime *plug_mgr_rt);

97
test/include/plugin_manager_gtest_mock.h
View File

@@ -1,97 +0,0 @@
#ifdef __cplusplus
extern "C"
{
#endif
#include "plugin_manager_interna.h"
#include "stellar_internal.h"
#include "stellar/session.h"
#include "lua_plugin_manage.h"
//mock stellar
struct stellar
{
struct plugin_manager_schema *plug_mgr;
struct lua_plugin_manage * lua_plug_mgr;
int thread_num;
};
struct packet
{
struct stellar *st;
enum packet_type type;
unsigned char ip_proto;
};
struct session
{
struct plugin_manager_runtime *plug_mgr_rt;
enum session_type type;
enum session_state state;
int sess_pkt_cnt;
};
enum session_state session_get_current_state(struct session *sess)
{
return sess->state;
}
enum session_type session_get_type(struct session *sess)
{
return sess->type;
}
int session_get_current_plugin_id(struct session *sess)
{
return sess->plug_mgr_rt->current_session_plugin_id;
}
struct plugin_manager_schema * stellar_plugin_manager_schema_get(struct stellar *st)
{
return st->plug_mgr;
}
int stellar_plugin_manager_schema_set(struct stellar *st, struct plugin_manager_schema *pm)
{
st->plug_mgr=pm;
return 0;
}
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunused-parameter"
int stellar_get_worker_thread_num(struct stellar *st)
{
return st->thread_num;
}
int stellar_get_current_thread_id(struct stellar *st)
{
return 3;
}
#pragma GCC diagnostic pop
struct stellar * packet_stellar_get(struct packet *pkt)
{
return pkt->st;
}
struct plugin_manager_runtime * session_plugin_manager_runtime_get(struct session *sess)
{
return sess->plug_mgr_rt;
}
unsigned char packet_get_ip_protocol(struct packet *pkt)
{
return pkt->ip_proto;
}
enum packet_type packet_get_type(const struct packet *pkt)
{
return pkt->type;
}
#ifdef __cplusplus
}
#endif

199
test/include/plugin_manager_interna.h
View File

@@ -1,199 +0,0 @@
#include "plugin_manager.h"
#include "stellar/stellar.h"
#include "stellar/session_exdata.h"
#include "stellar/session_mq.h"
#include "stellar/packet_exdata.h"
#include "stellar/packet_mq.h"
#include "bitmap.h"
#include "utarray.h"
struct per_thread_exdata_array
{
struct stellar_exdata *exdata_array;
};
struct stellar_message;
struct plugin_manger_per_thread_data
{
struct per_thread_exdata_array per_thread_pkt_exdata_array;
struct stellar_message *priority_mq[SESSION_MQ_PRIORITY_MAX];// message list
struct stellar_message *dealth_letter_queue;// dlq list
long long pub_packet_msg_cnt;
};
struct plugin_manager_schema
{
struct stellar *st;
UT_array *plugin_load_specs_array;
UT_array *packet_exdata_schema_array;
UT_array *session_exdata_schema_array;
UT_array *stellar_mq_schema_array;
UT_array *registered_session_plugin_array;
UT_array *registered_packet_plugin_array;
UT_array *registered_polling_plugin_array;
int packet_mq_topic_num;
int session_mq_topic_num;
int packet_topic_subscriber_num;
int session_topic_subscriber_num;
int tcp_topic_id;
int tcp_stream_topic_id;
int udp_topic_id;
int egress_topic_id;
int control_packet_topic_id;
int max_message_dispatch;
struct plugin_manger_per_thread_data *per_thread_data;
}__attribute__((aligned(sizeof(void*))));
enum plugin_exdata_state
{ INIT, ACTIVE, EXIT };
struct stellar_exdata
{
void *exdata;
enum plugin_exdata_state state;
};
struct stellar_exdata_schema
{
char *name;
union
{
void *free_func;
session_exdata_free *sess_free_func;
packet_exdata_free *pkt_free_func;
};
void *free_arg;
int idx;
}__attribute__((aligned(sizeof(void*))));
enum stellar_topic_type
{
ON_SESSION_TOPIC,
ON_PACKET_TOPIC,
};
struct stellar_message
{
struct
{
int topic_id;
enum stellar_topic_type type;
enum session_mq_priority priority;
} header;
void *body;
struct stellar_message *next, *prev;
} __attribute__((aligned(sizeof(void *))));
typedef struct stellar_mq_subscriber
{
int topic_subscriber_idx;
int plugin_idx;
union
{
on_session_msg_cb_func *sess_msg_cb;
on_packet_msg_cb_func *pkt_msg_cb;
void *msg_cb;
};
struct stellar_mq_subscriber *next, *prev;
}stellar_mq_subscriber __attribute__((aligned(sizeof(void*))));
struct stellar_mq_topic_schema
{
char *topic_name;
void *free_cb_arg;
int topic_id;
int subscriber_cnt;
int is_destroyed;
union
{
void *free_cb;
session_msg_free_cb_func *sess_msg_free_cb;
packet_msg_free_cb_func *pkt_msg_free_cb;
};
struct stellar_mq_subscriber *subscribers;
}__attribute__((aligned(sizeof(void*))));
struct session_plugin_ctx_runtime
{
enum plugin_exdata_state state;
int session_plugin_id;
void *plugin_ctx;
}__attribute__((aligned(sizeof(void*))));
struct plugin_manager_runtime
{
struct plugin_manager_schema *plug_mgr;
struct session *sess;
struct bitmap *session_mq_status; //N * M bits, N topic, M subscriber
struct bitmap *session_topic_status; //N bits, N topic
struct stellar_exdata *sess_exdata_array;
struct session_plugin_ctx_runtime *plugin_ctx_array;//N plugins TODO: call alloc and free
int current_session_plugin_id;
int pub_session_msg_cnt;
}__attribute__((aligned(sizeof(void*))));
struct registered_packet_plugin_schema
{
char ip_protocol;
plugin_on_packet_func *on_packet;
void *plugin_env;
UT_array *registed_packet_mq_subscriber_info;
}__attribute__((aligned(sizeof(void*))));
struct registered_polling_plugin_schema
{
plugin_on_polling_func *on_polling;
void *plugin_env;
}__attribute__((aligned(sizeof(void*))));
struct stellar_mq_subscriber_info
{
int topic_id;
int subscriber_idx;
}__attribute__((aligned(sizeof(void*))));
struct registered_session_plugin_schema
{
session_ctx_new_func *on_ctx_new;
session_ctx_free_func *on_ctx_free;
void *plugin_env;
UT_array *registed_session_mq_subscriber_info;
}__attribute__((aligned(sizeof(void*))));
#define SESSION_PULGIN_ID_BASE 0x00000
#define PACKET_PULGIN_ID_BASE 0x10000
#define POLLING_PULGIN_ID_BASE 0x20000
/*******************************
* PLUGIN MANAGER INIT & EXIT *
*******************************/
#define MAX_MSG_PER_DISPATCH 128
#include <dlfcn.h>
struct plugin_specific
{
char plugin_name[256];
plugin_on_load_func *load_cb;
plugin_on_unload_func *unload_cb;
void *plugin_ctx;
}__attribute__((aligned(sizeof(void*))));

27
test/include/stellar_internal.h
View File

@@ -1,27 +0,0 @@
#pragma once
#include "stellar/stellar.h"
struct plugin_manager_schema;
struct plugin_manager_runtime;
int stellar_plugin_manager_schema_set(struct stellar *st, struct plugin_manager_schema *pm);
struct plugin_manager_schema * stellar_plugin_manager_schema_get(struct stellar *st);
struct plugin_manager_runtime * session_plugin_manager_runtime_get(struct session *sess);
enum packet_type
{
UNKNOWN,
IPv4,
IPv6,
UDP,
TCP,
TCP_STREAM,
CONTROL,
};
enum packet_type packet_get_type(const struct packet *pkt);
struct stellar * packet_stellar_get(struct packet *pkt);

65
test/plugin/example_lua_plugin.lua
View File

@@ -1,65 +0,0 @@
function plugin_ctx_new(sess, plug_env, sess_context)
print("now lua plugin create sess ctx, plugin id", plug_env.id)
sess_context.count = 0
sess_context.payload_len = 0
msg = {}
msg.data = "this is lua plugin message"
session_mq.publish_message(sess, plug_env.topic_id, msg)
end
function plugin_ctx_free(sess, sess_context, plug_env)
print("now lua plugin begin to free sess ctx")
print("total packet count is", sess_context.count)
print("total paylaod length is", sess_context.payload_len)
end
function on_message_fn(sess, topic_id, msg, sess_context, env)
sess_context.count = sess_context.count + 1
print("lua plugin call on message fn", sess_context.count)
payload, payload_len = session.get_payload(sess)
sess_context.payload_len = sess_context.payload_len + payload_len
print(payload)
end
function free_message(sess, msg, msg_free_arg)
print("message need to free is", msg.data)
print("free arg data is", msg_free_arg.data)
end
function plugin_load(stellar, plug_env)
print("now begin to load lua example plugin")
plug_env.st = stellar
plug_env.id = session_plugin.register(stellar, plugin_ctx_new, plugin_ctx_free, plug_env)
plug_env.data = "this is lua example plug env data"
--[[ 获取TCP topic, 并完成订阅 ]]
tcp_topic_id = session_mq.get_topic_id(stellar, "TCP")
print("lua plugin get TCP topic id is", tcp_topic_id)
session_mq.subscribe_topic(stellar, tcp_topic_id, on_message_fn, plug_env.id)
plug_env.tcp_topic_id = tcp_topic_id
--[[ 创建TOPIC_SESSION_STAT, 并完成函数注册 ]]
msg_free_arg = {}
msg_free_arg.data = "this is lua plugin message free arg data"
test_topic_id = session_mq.get_topic_id(stellar, "LUA_TOPIC_SESSION_STAT")
if (test_topic_id < 0)
then
--[[ 该消息未创建, 创建该topic ]]
test_topic_id = session_mq.create_topic(stellar, "LUA_TOPIC_SESSION_STAT", free_message, msg_free_arg)
print("create topic is", test_topic_id)
else
--[[ 如果该消息已经注册, 更新其注册函数 ]]
session_mq.update_topic(stellar, test_topic_id, free_message, msg_free_arg)
print("topic already created, id is", test_topic_id)
end
plug_env.tcp_topic_id = tcp_topic_id
plug_env.topic_id = test_topic_id
end
function plugin_unload(plug_env)
print("now unload lua example plugin")
print("plugin env data is", plug_env.data)
print("now destory topic id is", plug_env.topic_id)
session_mq.destory_topic(plug_env.st, plug_env.topic_id)
end

11
test/plugin/example_plugin_load.lua
View File

@@ -1,11 +0,0 @@
function plugin_load(stellar, plug_env)
print("now load lua plugin example load")
print("this example aims to test load function")
plug_env.data = "load plugin env"
print("plugin env data is", plug_env.data)
end
function plugin_unload(plug_env)
print("now unload lua plugin example load")
print("plugin env data is", plug_env.data)
end

33
test/plugin/example_plugin_message.lua
View File

@@ -1,33 +0,0 @@
function plugin_ctx_new(sess, plug_env, sess_context)
-- print("now create new ctx example message, plugin id", plug_env.id)
end
function plugin_ctx_free(sess, sess_context, plug_env)
-- print("now begin to free ctx context example message")
end
function on_message(sess, topic_id, msg, sess_context, env)
-- print("message call on message function, id", topic_id)
print("get message is", topic_id, msg.data)
end
function plugin_load(stellar, plug_env)
plug_env.st = stellar
plug_env.id = session_plugin.register(stellar, plugin_ctx_new, plugin_ctx_free, plug_env)
plug_env.data = "this is message example plug env data"
--[[ 订阅TOPIC_SESSION_STAT, 并完成函数注册 ]]
test_topic_id = session_mq.get_topic_id(stellar, "LUA_TOPIC_SESSION_STAT")
if (test_topic_id < 0)
then
--[[ 该消息未创建, 创建该topic ]]
test_topic_id = session_mq.create_topic(stellar, "LUA_TOPIC_SESSION_STAT", nil, nil)
print("create topic is", test_topic_id)
end
session_mq.subscribe_topic(stellar, test_topic_id, on_message, plug_env.id)
end
function plugin_unload(plug_env)
print("now unload lua plugin example message")
print("plugin env data is", plug_env.data)
end

25
test/plugin/example_plugin_regist.lua
View File

@@ -1,25 +0,0 @@
function plugin_ctx_new(sess, plug_env, sess_context)
print("now begin to create new ctx context example regist")
sess_context.data = "this is example regist sess context data"
print("session data is", sess_context.data)
end
function plugin_ctx_free(sess, sess_context, plug_env)
print("now begin to free ctx context example regist")
print("session data is", sess_context.data)
end
function plugin_load(stellar, plug_env)
print("now load lua plugin example regist")
print("this example aims to test regist function")
plug_env.data = "my example regist plugin env"
--[[ 由于未订阅消息, 注册函数应该永远不会触发 ]]
plug_env.id = session_plugin.register(stellar, plugin_ctx_new, plugin_ctx_free, plug_env)
print("plugin env data is", plug_env.data)
print("after regist, plugin id is", plug_env.id)
end
function plugin_unload(plug_env)
print("now unload lua plugin example regist")
print("plugin env data is", plug_env.data)
end

65
test/plugin/example_plugin_topic.lua
View File

@@ -1,65 +0,0 @@
function plugin_ctx_new(sess, plug_env, sess_context)
print("now create new ctx example topic, plugin id", plug_env.id)
sess_context.count = 0
sess_context.payload_len = 0
msg = {}
msg.data = "this is message"
session_mq.publish_message(sess, plug_env.topic_id, msg)
end
function plugin_ctx_free(sess, sess_context, plug_env)
print("now begin to free ctx context example topic")
print("total packet count is", sess_context.count)
print("total paylaod length is", sess_context.payload_len)
end
function on_message_fn(sess, topic_id, msg, sess_context, env)
sess_context.count = sess_context.count + 1
-- print("topic call on message function, id", topic_id)
payload, payload_len = session.get_payload(sess)
sess_context.payload_len = sess_context.payload_len + payload_len
end
function free_message(sess, msg, msg_free_arg)
print("message need to free is", msg.data)
print("env id is ", msg_free_arg.id)
end
function plugin_load(stellar, plug_env)
print("now begin to load plugin example topic")
print("this example aims to test topic functions")
plug_env.st = stellar
plug_env.id = session_plugin.register(stellar, plugin_ctx_new, plugin_ctx_free, plug_env)
plug_env.data = "this is topic example plug env data"
--[[ 获取TCP topic, 并完成订阅 ]]
tcp_topic_id = session_mq.get_topic_id(stellar, "TCP")
print("get TCP topic id is", tcp_topic_id)
session_mq.subscribe_topic(stellar, tcp_topic_id, on_message_fn, plug_env.id)
plug_env.tcp_topic_id = tcp_topic_id
--[[ 创建TOPIC_SESSION_STAT, 并完成函数注册 ]]
msg_free_arg = {}
msg_free_arg.data = "this is example topic msg private data"
test_topic_id = session_mq.get_topic_id(stellar, "LUA_TOPIC_SESSION_STAT")
if (test_topic_id < 0)
then
--[[ 该消息未创建, 创建该topic ]]
test_topic_id = session_mq.create_topic(stellar, "LUA_TOPIC_SESSION_STAT", free_message, msg_free_arg)
print("create topic is", test_topic_id)
else
--[[ 如果该消息已经注册, 更新其注册函数 ]]
session_mq.update_topic(stellar, test_topic_id, free_message, msg_free_arg)
print("topic already created, id is", test_topic_id)
end
plug_env.tcp_topic_id = tcp_topic_id
plug_env.topic_id = test_topic_id
end
function plugin_unload(plug_env)
print("now unload lua plugin example topic")
print("plugin env data is", plug_env.data)
print("now destory topic id is", plug_env.topic_id)
session_mq.destory_topic(plug_env.st, plug_env.topic_id)
end

363
test/simple_example_plugin.c
View File

@@ -1,363 +0,0 @@
#pragma GCC diagnostic ignored "-Wunused-parameter"
#include "stellar/stellar.h"
#include "stellar/session.h"
#include "stellar/utils.h"
#include "stellar/session_exdata.h"
#include "stellar/session_mq.h"
#include "stellar/packet_exdata.h"
#include "stellar/packet_mq.h"
#include <stdio.h>
#include <string.h>
#include <assert.h>
struct simple_stellar_plugin_env
{
struct stellar *st;
int session_plugin_id;
int session_exdata_idx;
int packet_exdata_idx;
int packet_topic_id;
int stat_topic_id;
int egress_topic_id;
int tcp_topic_id;
int udp_topic_id;
int tcp_stream_topic_id;
long long tcp_packet_count;
long long udp_packet_count;
long long icmp_packet_count;
long long icmp6_packet_count;
long long polling_times;
};
struct mq_message_stat
{
uint32_t c2s_pkts;
uint32_t c2s_bytes;
uint32_t s2c_pkts;
uint32_t s2c_bytes;
uint32_t c2s_stream_pkts;
uint32_t c2s_stream_bytes;
uint32_t s2c_stream_pkts;
uint32_t s2c_stream_bytes;
};
static void print_session_stat(struct session *sess, struct mq_message_stat *stat, int plugin_id, const char *banner)
{
if (stat)
{
printf("%s(plug:%d)-----------%20s: ", banner, plugin_id, session_get0_readable_addr(sess));
printf("server-pkt=%u, server-count=%u, client-pkt=%u, client-count=%u, ", stat->c2s_pkts, stat->c2s_bytes,
stat->s2c_pkts, stat->s2c_bytes);
printf("total-pkt=%u, ", stat->c2s_pkts + stat->s2c_pkts);
printf("total-count=%u\n", stat->c2s_bytes + stat->s2c_bytes);
if(stat->c2s_stream_pkts+stat->s2c_stream_pkts > 0)
{
printf("----------------server-stream_pkt=%u, server-stream_count=%u, client-stream_pkt=%u, client-stream_count=%u\n", stat->c2s_stream_pkts, stat->c2s_stream_bytes,
stat->s2c_stream_pkts, stat->s2c_stream_bytes);
}
}
return;
}
/*******************************
* simple session & packet plugin *
*******************************/
static void *simple_plugin_session_ctx_new(struct session *sess, void *plugin_env)
{
struct mq_message_stat * stat= CALLOC(struct mq_message_stat, 1);
struct simple_stellar_plugin_env *env=(struct simple_stellar_plugin_env *)plugin_env;
session_exdata_set(sess, env->session_exdata_idx, stat);
return stat;
}
static void simple_plugin_session_ctx_free(struct session *sess, void *session_ctx, void *plugin_env)
{
struct simple_stellar_plugin_env *env=(struct simple_stellar_plugin_env *)plugin_env;
struct mq_message_stat *stat = (struct mq_message_stat *)session_ctx;
print_session_stat(sess, stat, env->session_plugin_id, __FUNCTION__);
session_exdata_set(sess, env->session_exdata_idx, NULL);
if(session_ctx)FREE(session_ctx);
return;
}
static void simple_plugin_on_session_msg(struct session *sess, int topic_id, const void *data, void *plugin_ctx, void *plugin_env)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
struct mq_message_stat *stat = (struct mq_message_stat *)plugin_ctx;
struct packet * pkt=(struct packet *)data;
if (pkt)
{
// TEST: try stellar_session_plugin_dettach_current_session when pkt > 3
if(stat->c2s_pkts+stat->s2c_pkts >= 3 && session_get_type(sess)== SESSION_TYPE_UDP)
{
stellar_session_plugin_dettach_current_session(sess);
return;
}
size_t payload_len = 0;
session_get0_current_payload(sess, &payload_len);
printf("get packet length is %lu\n", payload_len);
int dir = packet_get_direction(pkt);
if (dir==PACKET_DIRECTION_C2S)
{
if(topic_id==env->tcp_stream_topic_id)
{
stat->c2s_stream_bytes += payload_len;
stat->c2s_stream_pkts += 1;
}
else
{
stat->c2s_bytes += payload_len;
stat->c2s_pkts += 1;
}
}
if (dir==PACKET_DIRECTION_S2C)
{
if(topic_id==env->tcp_stream_topic_id)
{
stat->s2c_stream_bytes += payload_len;
stat->s2c_stream_pkts += 1;
}
else
{
stat->s2c_bytes += payload_len;
stat->s2c_pkts += 1;
}
}
session_mq_publish_message(sess, ((struct simple_stellar_plugin_env *)plugin_env)->stat_topic_id, stat);
}
return;
}
void simple_plugin_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
packet_exdata_set(pkt, env->packet_exdata_idx, env);
packet_mq_publish_message(pkt, env->packet_topic_id, env);
switch (ip_protocol)
{
case IPPROTO_TCP:
env->tcp_packet_count++;
break;
case IPPROTO_UDP:
env->udp_packet_count++;
break;
case IPPROTO_ICMP:
env->icmp_packet_count++;
break;
case IPPROTO_ICMPV6:
env->icmp6_packet_count++;
break;
default:
perror("invalid ip_protocol\n");
exit(-1);
break;
}
return;
}
void simple_plugin_packet_get_exdata(struct packet *pkt, unsigned char ip_protocol , void *plugin_env)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
struct simple_stellar_plugin_env *exdata = (struct simple_stellar_plugin_env *)packet_exdata_get(pkt, env->packet_exdata_idx);
if(memcmp(env, exdata, sizeof(struct simple_stellar_plugin_env)) != 0)
{
abort();
}
return;
}
int simple_plugin_on_polling(void *plugin_env)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
env->polling_times++;
return 0;
}
static void simple_plugin_packet_exdata_free(struct packet *pkt , int idx , void *ex_ptr, void *arg)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)arg;
assert(env);
struct simple_stellar_plugin_env *exdata = (struct simple_stellar_plugin_env *)ex_ptr;
if(memcmp(env, exdata, sizeof(struct simple_stellar_plugin_env)) != 0)
{
abort();
}
}
static void simple_plugin_packet_msg_free(struct packet *pkt , void *msg, void *msg_free_arg)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)msg_free_arg;
assert(env);
struct simple_stellar_plugin_env *exdata = (struct simple_stellar_plugin_env *)msg;
if(memcmp(env, exdata, sizeof(struct simple_stellar_plugin_env)) != 0)
{
abort();
}
}
static void simple_plugin_on_packet_msg_cb(struct packet *pkt , int topic_id , const void *msg, void *plugin_env)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
assert(env);
struct simple_stellar_plugin_env *exdata = (struct simple_stellar_plugin_env *)msg;
if(memcmp(env, exdata, sizeof(struct simple_stellar_plugin_env)) != 0)
{
abort();
}
}
void *simple_session_packet_plugin_init(struct stellar *st)
{
struct simple_stellar_plugin_env *env = CALLOC(struct simple_stellar_plugin_env, 1);
env->st = st;
env->session_exdata_idx = stellar_session_exdata_new_index(st, "EXDATA_SESSION_STAT", NULL, NULL);
env->session_plugin_id = stellar_session_plugin_register(st,
simple_plugin_session_ctx_new,
simple_plugin_session_ctx_free,
env);
int tcp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_TCP, simple_plugin_on_packet, env);
int udp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_UDP, simple_plugin_on_packet, env);
int icmp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_ICMP, simple_plugin_on_packet, env);
int icmp6_plugin_id=stellar_packet_plugin_register(st, IPPROTO_ICMPV6, simple_plugin_on_packet, env);
if(tcp_plugin_id < 0 || udp_plugin_id < 0 || icmp_plugin_id < 0 || icmp6_plugin_id < 0)
{
perror("register packet plugin return invalid plugin id\n");
exit(-1);
}
env->packet_topic_id=stellar_packet_mq_get_topic_id(st, "TOPIC_PACKET_ENV");
if(env->packet_topic_id < 0)
{
env->packet_topic_id=stellar_packet_mq_create_topic(st, "TOPIC_PACKET_ENV", simple_plugin_packet_msg_free, env);
}
tcp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_TCP, simple_plugin_packet_get_exdata, env);
udp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_UDP, simple_plugin_packet_get_exdata, env);
icmp_plugin_id=stellar_packet_plugin_register(st, IPPROTO_ICMP, simple_plugin_packet_get_exdata, env);
icmp6_plugin_id=stellar_packet_plugin_register(st, IPPROTO_ICMPV6, simple_plugin_packet_get_exdata, env);
if(tcp_plugin_id < 0 || udp_plugin_id < 0 || icmp_plugin_id < 0 || icmp6_plugin_id < 0)
{
perror("register packet plugin get exdata return invalid plugin id\n");
exit(-1);
}
stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, tcp_plugin_id);
stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, udp_plugin_id);
stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, icmp_plugin_id);
stellar_packet_mq_subscribe(st, env->packet_topic_id, simple_plugin_on_packet_msg_cb, icmp6_plugin_id);
env->packet_exdata_idx=stellar_packet_exdata_new_index(st, "EXDATA_PACKET_ENV", simple_plugin_packet_exdata_free, env);
int polling_plugin_id=stellar_polling_plugin_register(st, simple_plugin_on_polling, env);
if(polling_plugin_id < 0)
{
perror("register polling plugin return invalid plugin id \n");
exit(-1);
}
env->tcp_stream_topic_id=stellar_session_mq_get_topic_id(st, TOPIC_TCP_STREAM);
env->tcp_topic_id=stellar_session_mq_get_topic_id(st, TOPIC_TCP);
env->udp_topic_id=stellar_session_mq_get_topic_id(st, TOPIC_UDP);
if(env->tcp_topic_id < 0 || env->udp_topic_id < 0 || env->tcp_stream_topic_id < 0)
{
perror("get tcp or udp topic id failed\n");
exit(-1);
}
stellar_session_mq_subscribe(st, env->tcp_stream_topic_id, simple_plugin_on_session_msg, env->session_plugin_id);
stellar_session_mq_subscribe(st, env->tcp_topic_id, simple_plugin_on_session_msg, env->session_plugin_id);
stellar_session_mq_subscribe(st, env->udp_topic_id, simple_plugin_on_session_msg, env->session_plugin_id);
int stat_topic_id=stellar_session_mq_get_topic_id(st, "TOPIC_SESSION_STAT");
if(stat_topic_id < 0)
{
stat_topic_id=stellar_session_mq_create_topic(st, "TOPIC_SESSION_STAT", NULL, NULL);
}
env->stat_topic_id = stat_topic_id;
return env;
}
void simple_session_packet_plugin_exit(void *plugin_env)
{
if(plugin_env)
{
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
printf("(%s):tcp_packet_num:%lld, udp_packet_num:%lld, icmp_packet_num:%lld, icmp6_packet_num:%lld, polling_times:%lld\n", __FUNCTION__, env->tcp_packet_count, env->udp_packet_count, env->icmp_packet_count, env->icmp6_packet_count, env->polling_times);
FREE(plugin_env);
}
return;
}
/*******************************
* simple plugin sub session stat *
*******************************/
static void simple_plugin_sub_session_stat_on_msg(struct session *sess, int topic_id, const void *data, void *plugin_ctx , void *plugin_env)
{
printf("get message, %d\n", topic_id);
struct simple_stellar_plugin_env *env = (struct simple_stellar_plugin_env *)plugin_env;
if(topic_id == env->egress_topic_id)
{
session_mq_ignore_message(sess, topic_id, env->session_plugin_id);
}
#if 0
if (topic_id == env->stat_topic_id)
{
struct mq_message_stat *stat = (struct mq_message_stat *)data;
struct mq_message_stat *exdata_stat = (struct mq_message_stat *)session_exdata_get(sess, env->session_exdata_idx);
if (memcmp(exdata_stat, stat, sizeof(struct mq_message_stat)) != 0)
{
perror("exdata and mq data not equal\n");
exit(-1);
}
// print_session_stat(sess, stat, env->plugin_id, __FUNCTION__);
session_mq_unignore_message(sess, env->egress_topic_id, env->session_plugin_id);
}
#endif
return;
}
void *simple_plugin_sub_session_stat_init(struct stellar *st)
{
printf("init new plugin, regist\n");
struct simple_stellar_plugin_env *env = CALLOC(struct simple_stellar_plugin_env, 1);
env->st = st;
env->session_exdata_idx = stellar_session_exdata_new_index(st, "EXDATA_SESSION_STAT", NULL, NULL);
int topic_id=stellar_session_mq_get_topic_id(st, "TOPIC_SESSION_STAT");
if(topic_id < 0)
{
topic_id=stellar_session_mq_create_topic(st, "TOPIC_SESSION_STAT", NULL, NULL);
}
env->stat_topic_id = topic_id;
env->session_plugin_id = stellar_session_plugin_register(st,
NULL,
NULL,
env);
stellar_session_mq_subscribe(st, topic_id, simple_plugin_sub_session_stat_on_msg, env->session_plugin_id);
// TEST: subscirbe egress message then ignore
env->egress_topic_id=stellar_session_mq_get_topic_id(st, TOPIC_EGRESS);
if(env->egress_topic_id < 0)
{
perror("get egress topic id failed\n");
exit(-1);
}
stellar_session_mq_subscribe(st, env->egress_topic_id, simple_plugin_sub_session_stat_on_msg, env->session_plugin_id);
return env;
}
void simple_plugin_sub_session_stat_exit(void *plugin_env)
{
if(plugin_env)FREE(plugin_env);
return;
}