GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-18 13:41:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1277 from eeff/eeff/dev

Browse Source 
feat: add function neu_adapter_rename
This commit is contained in:
fengzeroz
2023-05-19 16:51:02 +08:00
committed by GitHub
parent 0815456cde 0f57174819
commit 233fe1978f
5 changed files with 108 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+67 -7
View File
@@ -89,18 +89,31 @@ static const adapter_callbacks_t callback_funs = {
REGISTER_METRIC(adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 0); \
REGISTER_METRIC(adapter, NEU_METRIC_RECV_MSGS_TOTAL, 0);
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
static inline void start_log_level_timer(neu_adapter_t *adapter)
{
int rv = 0;
neu_adapter_t * adapter = NULL;
neu_event_io_param_t param = { 0 };
neu_event_timer_param_t timer_level = {
.second = 30,
.millisecond = 0,
.cb = level_check,
.usr_data = adapter,
.type = NEU_EVENT_TIMER_BLOCK,
};
adapter->timer_lev = neu_event_add_timer(adapter->events, timer_level);
}
static inline void stop_log_level_timer(neu_adapter_t *adapter)
{
neu_event_del_timer(adapter->events, adapter->timer_lev);
adapter->timer_lev = NULL;
}
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
{
int rv = 0;
neu_adapter_t * adapter = NULL;
neu_event_io_param_t param = { 0 };
switch (info->module->type) {
case NEU_NA_TYPE_DRIVER:
adapter = (neu_adapter_t *) neu_adapter_driver_create();
@@ -178,8 +191,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
rv = nng_dial(adapter->sock, neu_manager_get_url(), &adapter->dialer, 0);
assert(rv == 0);
timer_level.usr_data = (void *) adapter;
adapter->timer_lev = neu_event_add_timer(adapter->events, timer_level);
start_log_level_timer(adapter);
nlog_notice("Success to create adapter: %s", adapter->name);
@@ -187,6 +199,49 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
return adapter;
}
int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
{
char *name = strdup(new_name);
if (NULL == name) {
return NEU_ERR_EINTERNAL;
}
zlog_category_t *log = zlog_get_category(name);
if (NULL == log) {
free(name);
return NEU_ERR_EINTERNAL;
}
stop_log_level_timer(adapter);
if (NEU_NA_TYPE_DRIVER == adapter->module->type) {
neu_adapter_driver_stop_group_timer((neu_adapter_driver_t *) adapter);
}
// fix metrics
if (adapter->metrics) {
neu_metrics_del_node(adapter);
}
free(adapter->name);
adapter->name = name;
adapter->metrics->name = name;
if (adapter->metrics) {
neu_metrics_add_node(adapter);
}
// fix log
neu_plugin_common_t *common = neu_plugin_to_plugin_common(adapter->plugin);
common->log = log;
strcpy(common->name, adapter->name);
zlog_level_switch(common->log, default_log_level);
start_log_level_timer(adapter);
if (NEU_NA_TYPE_DRIVER == adapter->module->type) {
neu_adapter_driver_start_group_timer((neu_adapter_driver_t *) adapter);
}
return 0;
}
void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
{
neu_reqresp_head_t header = { .type = NEU_REQ_NODE_INIT };
@@ -797,6 +852,11 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
if (NULL != adapter->metrics) {
neu_metrics_del_node(adapter);
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, adapter->metrics->entries, e)
{
neu_metrics_unregister_entry(e->name);
}
neu_node_metrics_free(adapter->metrics);
}
@@ -824,7 +884,7 @@ int neu_adapter_uninit(neu_adapter_t *adapter)
neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter);
}
neu_event_del_timer(adapter->events, adapter->timer_lev);
stop_log_level_timer(adapter);
nlog_notice("Stop the adapter(%s)", adapter->name);
return 0;
src/adapter/adapter_internal.h
+2
View File
@@ -66,6 +66,8 @@ typedef struct adapter_msg_handler {
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
void neu_adapter_init(neu_adapter_t *adapter, bool auto_start);
int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name);
int neu_adapter_start(neu_adapter_t *adapter);
int neu_adapter_start_single(neu_adapter_t *adapter);
int neu_adapter_stop(neu_adapter_t *adapter);
src/adapter/driver/driver.c
+36
View File
@@ -206,6 +206,42 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
return 0;
}
void neu_adapter_driver_start_group_timer(neu_adapter_driver_t *driver)
{
group_t *el = NULL, *tmp = NULL;
HASH_ITER(hh, driver->groups, el, tmp)
{
uint32_t interval = neu_group_get_interval(el->group);
neu_event_timer_param_t param = {
.second = interval / 1000,
.millisecond = interval % 1000,
.usr_data = el,
.type = NEU_EVENT_TIMER_NOBLOCK,
};
param.cb = report_callback;
el->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
param.type = driver->adapter.module->timer_type;
param.cb = read_callback;
el->read = neu_event_add_timer(driver->driver_events, param);
}
}
void neu_adapter_driver_stop_group_timer(neu_adapter_driver_t *driver)
{
group_t *el = NULL, *tmp = NULL;
HASH_ITER(hh, driver->groups, el, tmp)
{
neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
el->report = NULL;
neu_event_del_timer(driver->driver_events, el->read);
el->read = NULL;
}
}
void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
src/adapter/driver/driver_internal.h
+3
View File
@@ -30,6 +30,9 @@ int neu_adapter_driver_stop(neu_adapter_driver_t *driver);
int neu_adapter_driver_init(neu_adapter_driver_t *driver);
int neu_adapter_driver_uninit(neu_adapter_driver_t *driver);
void neu_adapter_driver_start_group_timer(neu_adapter_driver_t *driver);
void neu_adapter_driver_stop_group_timer(neu_adapter_driver_t *driver);
void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req);
src/base/metrics.c
-5
View File
@@ -244,11 +244,6 @@ void neu_metrics_del_node(const neu_adapter_t *adapter)
{
pthread_rwlock_wrlock(&g_metrics_mtx_);
HASH_DEL(g_metrics_.node_metrics, adapter->metrics);
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, adapter->metrics->entries, e)
{
metrics_unregister_entry(e->name);
}
pthread_rwlock_unlock(&g_metrics_mtx_);
}