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

🦄 refactor(stellar_module to module): simplify stellar module to module

Browse Source
This commit is contained in:
yangwei
2024-11-05 09:39:10 +08:00
parent a415794428
commit 7f81e46522
17 changed files with 274 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
infra/module_manager/module_manager.c
View File

@@ -1,6 +1,5 @@
#include "module_manager_interna.h"
#include "stellar/module_manager.h"
#include "stellar/utils.h"
#include <dlfcn.h>
#include <stdbool.h>
@@ -14,10 +13,10 @@
#include "toml/toml.h"
struct stellar_module_manager *stellar_module_manager_new(const char *module_spec_toml_path, int max_thread_num, struct mq_schema *mq_schema, struct logger *logger)
struct module_manager *module_manager_new(const char *module_spec_toml_path, int max_thread_num, struct mq_schema *mq_schema, struct logger *logger)
{
struct stellar_module_manager *mod_mgr = CALLOC(struct stellar_module_manager, 1);
struct module_manager *mod_mgr = CALLOC(struct module_manager, 1);
mod_mgr->schema.max_thread_num=max_thread_num;
mod_mgr->schema.mq_schema=mq_schema;
mod_mgr->schema.logger=logger;
@@ -63,7 +62,7 @@ struct stellar_module_manager *stellar_module_manager_new(const char *module_spe
if (mod_mgr->module_specs[i].on_instance_init_cb)
{
mod_mgr->module_specs[i].mod = mod_mgr->module_specs[i].on_instance_init_cb(mod_mgr);
if (stellar_module_manager_get_module(mod_mgr, mod_mgr->module_specs[i].mod->name) == NULL)
if (module_manager_get_module(mod_mgr, mod_mgr->module_specs[i].mod->name) == NULL)
{
mod_mgr->module_specs[i].init_succ = true;
}
@@ -122,7 +121,7 @@ MODULE_SPEC_LOAD_END:
}
void stellar_module_manager_free(struct stellar_module_manager *mod_mgr)
void module_manager_free(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return;
if(mod_mgr->module_spec_toml_path)FREE(mod_mgr->module_spec_toml_path);
@@ -147,25 +146,25 @@ void stellar_module_manager_free(struct stellar_module_manager *mod_mgr)
return;
}
int stellar_module_manager_get_max_thread_num(struct stellar_module_manager*mod_mgr)
int module_manager_get_max_thread_num(struct module_manager*mod_mgr)
{
if(mod_mgr==NULL)return -1;
return mod_mgr->schema.max_thread_num;
}
struct mq_schema *stellar_module_manager_get_mq_schema(struct stellar_module_manager *mod_mgr)
struct mq_schema *module_manager_get_mq_schema(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return NULL;
return mod_mgr->schema.mq_schema;
}
struct logger *stellar_module_manager_get_logger(struct stellar_module_manager *mod_mgr)
struct logger *module_manager_get_logger(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return NULL;
return mod_mgr->schema.logger;
}
const char *stellar_module_manager_get_toml_path(struct stellar_module_manager *mod_mgr)
const char *module_manager_get_toml_path(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return NULL;
return mod_mgr->module_spec_toml_path;
@@ -174,17 +173,17 @@ const char *stellar_module_manager_get_toml_path(struct stellar_module_manager *
__thread int local_thread_id=-1;
__thread struct mq_runtime *local_mq_rt=NULL;
int stellar_module_manager_get_thread_id(struct stellar_module_manager* mod_mgr __unused)
int module_manager_get_thread_id(struct module_manager* mod_mgr __unused)
{
return local_thread_id;
}
struct mq_runtime *stellar_module_manager_get_mq_runtime(struct stellar_module_manager *mod_mgr __unused)
struct mq_runtime *module_manager_get_mq_runtime(struct module_manager *mod_mgr __unused)
{
return local_mq_rt;
}
void stellar_module_manager_register_thread(struct stellar_module_manager* mod_mgr, int thread_id, struct mq_runtime *mq_rt)
void module_manager_register_thread(struct module_manager* mod_mgr, int thread_id, struct mq_runtime *mq_rt)
{
local_thread_id=thread_id;
local_mq_rt=mq_rt;
@@ -200,7 +199,7 @@ void stellar_module_manager_register_thread(struct stellar_module_manager* mod_m
return;
}
void stellar_module_manager_unregister_thread(struct stellar_module_manager *mod_mgr, int thread_id)
void module_manager_unregister_thread(struct module_manager *mod_mgr, int thread_id)
{
assert(local_thread_id==thread_id);
for(int i=0; i<mod_mgr->load_module_num; i++)
@@ -216,7 +215,7 @@ void stellar_module_manager_unregister_thread(struct stellar_module_manager *mod
return;
}
struct stellar_module *stellar_module_manager_get_module(struct stellar_module_manager *mod_mgr, const char *module_name)
struct module *module_manager_get_module(struct module_manager *mod_mgr, const char *module_name)
{
if(mod_mgr==NULL || module_name == NULL)return NULL;
if (mod_mgr->module_specs)
@@ -238,41 +237,41 @@ struct stellar_module *stellar_module_manager_get_module(struct stellar_module_m
*******************************************/
struct stellar_module *stellar_module_new(const char *name, void *ctx)
struct module *module_new(const char *name, void *ctx)
{
struct stellar_module *mod = CALLOC(struct stellar_module, 1);
struct module *mod = CALLOC(struct module, 1);
memcpy(mod->name, name, MIN(NAME_MAX, strlen(name)));
mod->module_ctx=ctx;
return mod;
}
void stellar_module_free(struct stellar_module *mod)
void module_free(struct module *mod)
{
if(mod==NULL)return;
FREE(mod);
return;
}
void * stellar_module_get_ctx(struct stellar_module *mod)
void * module_get_ctx(struct module *mod)
{
if(mod==NULL)return NULL;
return mod->module_ctx;
}
void stellar_module_set_ctx(struct stellar_module *mod, void *ctx)
void module_set_ctx(struct module *mod, void *ctx)
{
if(mod==NULL)return;
mod->module_ctx=ctx;
return;
}
const char *stellar_module_get_name(struct stellar_module* mod)
const char *module_get_name(struct module* mod)
{
if(mod==NULL)return NULL;
return mod->name;
}
void stellar_module_set_name(struct stellar_module* mod, const char *name)
void module_set_name(struct module* mod, const char *name)
{
if(mod==NULL)return;
memcpy(mod->name, name, MIN(NAME_MAX, strlen(name)));
@@ -293,15 +292,15 @@ static void on_polling_dispatch(int topic_id __unused,
void *on_msg_cb_arg,
void *dispatch_arg)
{
struct stellar_module_manager *mod_mgr=(struct stellar_module_manager *)dispatch_arg;
struct module_manager *mod_mgr=(struct module_manager *)dispatch_arg;
module_on_polling_func *polling = (module_on_polling_func *)on_msg_cb;
polling(mod_mgr, on_msg_cb_arg);
}
int stellar_module_manager_polling_subscribe(struct stellar_module_manager *mod_mgr, module_on_polling_func on_polling, void *polling_arg)
int module_manager_polling_subscribe(struct module_manager *mod_mgr, module_on_polling_func on_polling, void *polling_arg)
{
if(mod_mgr == NULL)return -1;
mod_mgr->topic_polling_id=mq_schema_get_topic_id(stellar_module_manager_get_mq_schema(mod_mgr), TOPIC_POLLING);
mod_mgr->topic_polling_id=mq_schema_get_topic_id(module_manager_get_mq_schema(mod_mgr), TOPIC_POLLING);
if(mod_mgr->topic_polling_id<0)
{
mod_mgr->topic_polling_id=mq_schema_create_topic(mod_mgr->schema.mq_schema, TOPIC_POLLING, on_polling_dispatch, mod_mgr, NULL, NULL);
@@ -311,17 +310,17 @@ int stellar_module_manager_polling_subscribe(struct stellar_module_manager *mod_
#pragma GCC diagnostic pop
void stellar_module_manager_polling_active(struct stellar_module_manager *mod_mgr)
void module_manager_polling_active(struct module_manager *mod_mgr)
{
if(mod_mgr == NULL)return;
mq_runtime_publish_message(local_mq_rt, mod_mgr->topic_polling_id, NULL);
}
void stellar_polling_dispatch(struct stellar_module_manager *mod_mgr)
void stellar_polling_dispatch(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return;
stellar_module_manager_polling_active(mod_mgr);
module_manager_polling_active(mod_mgr);
mq_runtime_dispatch(local_mq_rt);
return;
}

10
infra/module_manager/module_manager_interna.h
View File

@@ -5,7 +5,7 @@ extern "C"
{
#endif
#include "stellar/module_manager.h"
#include "stellar/module.h"
#include "stellar/mq.h"
@@ -13,7 +13,7 @@ extern "C"
#include <stdbool.h>
struct stellar_module
struct module
{
char name[NAME_MAX];
void *module_ctx;
@@ -21,7 +21,7 @@ struct stellar_module
struct module_spec_load
{
struct stellar_module *mod;
struct module *mod;
module_on_instance_init_func *on_instance_init_cb;
module_on_instance_exit_func *on_instance_exit_cb;
module_on_thread_init_func *on_thread_init_cb;
@@ -35,7 +35,7 @@ struct module_spec_load
}__attribute__((aligned(sizeof(void*))));
struct stellar_module_manager
struct module_manager
{
char *module_spec_toml_path;
struct module_spec_load *module_specs;
@@ -50,7 +50,7 @@ struct stellar_module_manager
}__attribute__((aligned(sizeof(void*))));
void stellar_polling_dispatch(struct stellar_module_manager *mod_mgr);
void stellar_polling_dispatch(struct module_manager *mod_mgr);
#ifdef __cplusplus
}

162
infra/module_manager/test/gtest_module_manager_main.cpp
View File

@@ -13,8 +13,8 @@
* TEST MODUEL MANAGER INTERNAL API *
***********************************/
extern "C" struct stellar_module *gtest_mock_init(struct stellar_module_manager *mod_mgr){return NULL;}
extern "C" void gtest_mock_exit(struct stellar_module_manager *mod_mgr, struct stellar_module *mod){}
extern "C" struct module *gtest_mock_init(struct module_manager *mod_mgr){return NULL;}
extern "C" void gtest_mock_exit(struct module_manager *mod_mgr, struct module *mod){}
const char *gtest_mock_spec_toml =
"[[module]]\n"
"path = \"\"\n"
@@ -31,18 +31,18 @@ TEST(module_manager_internal, stellar_module_manager_new_with_toml) {
write(fd, gtest_mock_spec_toml, strlen(gtest_mock_spec_toml));
close(fd);
struct stellar_module_manager *mod_mgr=stellar_module_manager_new(toml_template, 10, mq_schema, NULL);
struct module_manager *mod_mgr=module_manager_new(toml_template, 10, mq_schema, NULL);
EXPECT_TRUE(mod_mgr!=NULL);
EXPECT_TRUE(stellar_module_manager_get_module(mod_mgr, "test")==NULL);
EXPECT_EQ(stellar_module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(stellar_module_manager_get_mq_schema(mod_mgr), mq_schema);
EXPECT_STREQ(stellar_module_manager_get_toml_path(mod_mgr), toml_template);
EXPECT_TRUE(module_manager_get_module(mod_mgr, "test")==NULL);
EXPECT_EQ(module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(module_manager_get_mq_schema(mod_mgr), mq_schema);
EXPECT_STREQ(module_manager_get_toml_path(mod_mgr), toml_template);
EXPECT_EQ(stellar_module_manager_get_thread_id(mod_mgr), -1);// no thread registered
EXPECT_TRUE(stellar_module_manager_get_mq_runtime(mod_mgr)==NULL);
EXPECT_EQ(module_manager_get_thread_id(mod_mgr), -1);// no thread registered
EXPECT_TRUE(module_manager_get_mq_runtime(mod_mgr)==NULL);
stellar_module_manager_free(mod_mgr);
module_manager_free(mod_mgr);
unlink(toml_template);
}
@@ -53,16 +53,16 @@ TEST(module_manager_internal, stellar_module_manager_new_with_toml) {
TEST(stellar_module, basic_new_and_free) {
struct stellar_module *mod = stellar_module_new("test", NULL);
struct module *mod = module_new("test", NULL);
EXPECT_TRUE(mod!=NULL);
stellar_module_set_name(mod, "test1");
EXPECT_STREQ(stellar_module_get_name(mod), "test1");
module_set_name(mod, "test1");
EXPECT_STREQ(module_get_name(mod), "test1");
stellar_module_set_ctx(mod, (void*)1);
EXPECT_EQ((long)stellar_module_get_ctx(mod), 1);
module_set_ctx(mod, (void*)1);
EXPECT_EQ((long)module_get_ctx(mod), 1);
stellar_module_free(mod);
module_free(mod);
}
/***********************************
@@ -72,57 +72,57 @@ TEST(stellar_module, basic_new_and_free) {
TEST(stellar_module_manager, new_with_null_toml) {
struct mq_schema *mq_schema=NULL;
struct stellar_module_manager *mod_mgr = stellar_module_manager_new(NULL, 10, mq_schema, NULL);
struct module_manager *mod_mgr = module_manager_new(NULL, 10, mq_schema, NULL);
EXPECT_TRUE(mod_mgr!=NULL);
EXPECT_TRUE(stellar_module_manager_get_module(mod_mgr, "test")==NULL);
EXPECT_EQ(stellar_module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(stellar_module_manager_get_mq_schema(mod_mgr), mq_schema);
EXPECT_TRUE(module_manager_get_module(mod_mgr, "test")==NULL);
EXPECT_EQ(module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(module_manager_get_mq_schema(mod_mgr), mq_schema);
EXPECT_TRUE(stellar_module_manager_get_mq_runtime(mod_mgr)==NULL);
EXPECT_EQ(stellar_module_manager_get_thread_id(mod_mgr), -1);// no thread registered
EXPECT_TRUE(module_manager_get_mq_runtime(mod_mgr)==NULL);
EXPECT_EQ(module_manager_get_thread_id(mod_mgr), -1);// no thread registered
stellar_module_manager_free(mod_mgr);
module_manager_free(mod_mgr);
}
TEST(stellar_module_manager, new_with_empty_toml) {
struct mq_schema *mq_schema=NULL;
struct stellar_module_manager *mod_mgr = stellar_module_manager_new("/dev/null", 10, mq_schema, NULL);
struct module_manager *mod_mgr = module_manager_new("/dev/null", 10, mq_schema, NULL);
EXPECT_TRUE(mod_mgr!=NULL);
EXPECT_TRUE(stellar_module_manager_get_module(mod_mgr, "test")==NULL);
EXPECT_EQ(stellar_module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(stellar_module_manager_get_mq_schema(mod_mgr), mq_schema);
EXPECT_TRUE(module_manager_get_module(mod_mgr, "test")==NULL);
EXPECT_EQ(module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(module_manager_get_mq_schema(mod_mgr), mq_schema);
EXPECT_EQ(stellar_module_manager_get_thread_id(mod_mgr), -1);// no thread registered
EXPECT_TRUE(stellar_module_manager_get_mq_runtime(mod_mgr)==NULL);
EXPECT_EQ(module_manager_get_thread_id(mod_mgr), -1);// no thread registered
EXPECT_TRUE(module_manager_get_mq_runtime(mod_mgr)==NULL);
stellar_module_manager_free(mod_mgr);
module_manager_free(mod_mgr);
}
TEST(stellar_module_manager, register_thread) {
struct mq_schema *mq_schema=(struct mq_schema*)1;
struct stellar_module_manager *mod_mgr=stellar_module_manager_new(NULL, 10, mq_schema, NULL);
struct module_manager *mod_mgr=module_manager_new(NULL, 10, mq_schema, NULL);
EXPECT_TRUE(mod_mgr!=NULL);
EXPECT_EQ((long)stellar_module_manager_get_mq_schema(mod_mgr), 1);
EXPECT_EQ((long)module_manager_get_mq_schema(mod_mgr), 1);
EXPECT_EQ(stellar_module_manager_get_thread_id(mod_mgr), -1);// no thread registered
EXPECT_TRUE(stellar_module_manager_get_mq_runtime(mod_mgr)==NULL);
EXPECT_EQ(module_manager_get_thread_id(mod_mgr), -1);// no thread registered
EXPECT_TRUE(module_manager_get_mq_runtime(mod_mgr)==NULL);
struct mq_runtime *mq_rt = (struct mq_runtime*)2;
stellar_module_manager_register_thread(mod_mgr, 1, mq_rt);
module_manager_register_thread(mod_mgr, 1, mq_rt);
EXPECT_EQ(stellar_module_manager_get_thread_id(mod_mgr), 1);
EXPECT_EQ((long)stellar_module_manager_get_mq_runtime(mod_mgr), 2);
EXPECT_EQ(module_manager_get_thread_id(mod_mgr), 1);
EXPECT_EQ((long)module_manager_get_mq_runtime(mod_mgr), 2);
stellar_module_manager_unregister_thread(mod_mgr, 1);
module_manager_unregister_thread(mod_mgr, 1);
EXPECT_EQ(stellar_module_manager_get_thread_id(mod_mgr), -1);
EXPECT_EQ((long)stellar_module_manager_get_mq_runtime(mod_mgr), 0);
EXPECT_EQ(module_manager_get_thread_id(mod_mgr), -1);
EXPECT_EQ((long)module_manager_get_mq_runtime(mod_mgr), 0);
stellar_module_manager_free(mod_mgr);
module_manager_free(mod_mgr);
}
@@ -130,38 +130,38 @@ TEST(stellar_module_manager, register_thread) {
* TEST MODULE MANAGER API *
***********************************/
extern "C" struct stellar_module *gtest_module_init(struct stellar_module_manager *mod_mgr)
extern "C" struct module *gtest_module_init(struct module_manager *mod_mgr)
{
struct stellar_module *mod = stellar_module_new("gtest", NULL);
EXPECT_STREQ(stellar_module_get_name(mod), "gtest");
stellar_module_set_ctx(mod, (void*)1);
struct module *mod = module_new("gtest", NULL);
EXPECT_STREQ(module_get_name(mod), "gtest");
module_set_ctx(mod, (void*)1);
return mod;
}
extern "C" void gtest_module_exit(struct stellar_module_manager *mod_mgr, struct stellar_module *mod)
extern "C" void gtest_module_exit(struct module_manager *mod_mgr, struct module *mod)
{
EXPECT_STREQ(stellar_module_get_name(mod), "gtest");
EXPECT_EQ((long)stellar_module_get_ctx(mod), 1);
EXPECT_STREQ(module_get_name(mod), "gtest");
EXPECT_EQ((long)module_get_ctx(mod), 1);
EXPECT_EQ(stellar_module_manager_get_module(mod_mgr, "gtest"), mod);
EXPECT_EQ(module_manager_get_module(mod_mgr, "gtest"), mod);
EXPECT_EQ(stellar_module_manager_get_thread_id(mod_mgr), -1);
EXPECT_EQ((long)stellar_module_manager_get_mq_runtime(mod_mgr), 0);
EXPECT_EQ(module_manager_get_thread_id(mod_mgr), -1);
EXPECT_EQ((long)module_manager_get_mq_runtime(mod_mgr), 0);
stellar_module_free(mod);
module_free(mod);
}
extern "C" void gtest_thread_init(struct stellar_module_manager *mod_mgr, int thread_id, struct stellar_module *mod)
extern "C" void gtest_thread_init(struct module_manager *mod_mgr, int thread_id, struct module *mod)
{
EXPECT_STREQ(stellar_module_get_name(mod), "gtest");
EXPECT_EQ((long)stellar_module_get_ctx(mod), 1);
EXPECT_STREQ(module_get_name(mod), "gtest");
EXPECT_EQ((long)module_get_ctx(mod), 1);
}
extern "C" void gtest_thread_exit(struct stellar_module_manager *mod_mgr, int thread_id, struct stellar_module *mod)
extern "C" void gtest_thread_exit(struct module_manager *mod_mgr, int thread_id, struct module *mod)
{
EXPECT_STREQ(stellar_module_get_name(mod), "gtest");
EXPECT_EQ((long)stellar_module_get_ctx(mod), 1);
EXPECT_STREQ(module_get_name(mod), "gtest");
EXPECT_EQ((long)module_get_ctx(mod), 1);
}
const char *gtest_module_spec_toml =
@@ -182,26 +182,26 @@ TEST(module_manager, basic_module) {
write(fd, gtest_module_spec_toml, strlen(gtest_module_spec_toml));
close(fd);
struct stellar_module_manager *mod_mgr=stellar_module_manager_new(toml_template, 10, mq_schema, NULL);
struct module_manager *mod_mgr=module_manager_new(toml_template, 10, mq_schema, NULL);
EXPECT_TRUE(mod_mgr!=NULL);
EXPECT_TRUE(stellar_module_manager_get_module(mod_mgr, "gtest")!=NULL);
EXPECT_TRUE(module_manager_get_module(mod_mgr, "gtest")!=NULL);
EXPECT_EQ(stellar_module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ((long)stellar_module_manager_get_mq_schema(mod_mgr), 1);
EXPECT_STREQ(stellar_module_manager_get_toml_path(mod_mgr), toml_template);
EXPECT_EQ(module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ((long)module_manager_get_mq_schema(mod_mgr), 1);
EXPECT_STREQ(module_manager_get_toml_path(mod_mgr), toml_template);
struct mq_runtime *mq_rt = (struct mq_runtime*)2;
stellar_module_manager_register_thread(mod_mgr, 1, mq_rt);
module_manager_register_thread(mod_mgr, 1, mq_rt);
EXPECT_EQ((long)stellar_module_manager_get_thread_id(mod_mgr), 1);
EXPECT_EQ((long)stellar_module_manager_get_mq_runtime(mod_mgr), 2);
EXPECT_EQ((long)module_manager_get_thread_id(mod_mgr), 1);
EXPECT_EQ((long)module_manager_get_mq_runtime(mod_mgr), 2);
stellar_module_manager_unregister_thread(mod_mgr, 1);
EXPECT_EQ((long)stellar_module_manager_get_thread_id(mod_mgr), -1);
EXPECT_EQ((long)stellar_module_manager_get_mq_runtime(mod_mgr), 0);
module_manager_unregister_thread(mod_mgr, 1);
EXPECT_EQ((long)module_manager_get_thread_id(mod_mgr), -1);
EXPECT_EQ((long)module_manager_get_mq_runtime(mod_mgr), 0);
stellar_module_manager_free(mod_mgr);
module_manager_free(mod_mgr);
unlink(toml_template);
}
@@ -216,13 +216,13 @@ struct test_module_polling_env
int polling_active_count;
};
void test_module_on_polling(struct stellar_module_manager* mod_mgr, void *polling_arg)
void test_module_on_polling(struct module_manager* mod_mgr, void *polling_arg)
{
struct test_module_polling_env *env = (struct test_module_polling_env*)polling_arg;
env->polling_count++;
if(env->polling_count%2==0)
{
stellar_module_manager_polling_active(mod_mgr);
module_manager_polling_active(mod_mgr);
env->polling_active_count++;
}
}
@@ -232,34 +232,34 @@ TEST(module_manager, basic_polling_module) {
struct mq_schema *mq_schema=mq_schema_new();
struct stellar_module_manager *mod_mgr=stellar_module_manager_new(NULL, 10, mq_schema, NULL);
struct module_manager *mod_mgr=module_manager_new(NULL, 10, mq_schema, NULL);
EXPECT_TRUE(mod_mgr!=NULL);
EXPECT_EQ(stellar_module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(stellar_module_manager_get_mq_schema(mod_mgr), mq_schema);
EXPECT_EQ(module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(module_manager_get_mq_schema(mod_mgr), mq_schema);
struct test_module_polling_env env={};
env.N_round=10;
stellar_module_manager_polling_subscribe(mod_mgr, test_module_on_polling, &env);
module_manager_polling_subscribe(mod_mgr, test_module_on_polling, &env);
struct mq_runtime *mq_rt = mq_runtime_new(mq_schema);
stellar_module_manager_register_thread(mod_mgr, 1, mq_rt);
module_manager_register_thread(mod_mgr, 1, mq_rt);
EXPECT_EQ((long)stellar_module_manager_get_thread_id(mod_mgr), 1);
EXPECT_EQ(stellar_module_manager_get_mq_runtime(mod_mgr), mq_rt);
EXPECT_EQ((long)module_manager_get_thread_id(mod_mgr), 1);
EXPECT_EQ(module_manager_get_mq_runtime(mod_mgr), mq_rt);
for(int i=0; i<env.N_round; i++)
{
stellar_polling_dispatch(mod_mgr);
}
stellar_module_manager_unregister_thread(mod_mgr, 1);
module_manager_unregister_thread(mod_mgr, 1);
mq_runtime_free(mq_rt);
mq_schema_free(mq_schema);
stellar_module_manager_free(mod_mgr);
module_manager_free(mod_mgr);
EXPECT_EQ(env.polling_count, env.N_round+env.polling_active_count);