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 #2437 from hxy7yx/main-2

Browse Source 
fix(datalayers):restore connection
This commit is contained in:
fengzeroz
2025-05-07 17:05:47 +08:00
committed by GitHub
parent e38df4cd90 a999c9f80c
commit fdfd008858
5 changed files with 47 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/datalayers/datalayers_handle.c
+8
View File
@@ -110,6 +110,7 @@ db_write_task_t *task_queue_pop(neu_plugin_t *plugin, task_queue_t *queue)
}
queue->size--;
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_CACHED_QUEUE_SIZE_METRIC(plugin, plugin->task_queue.size);
return task;
@@ -340,11 +341,15 @@ int handle_trans_data(neu_plugin_t * plugin,
{
int rv = 0;
pthread_mutex_lock(&plugin->plugin_mutex);
if (NULL == plugin->client) {
pthread_mutex_unlock(&plugin->plugin_mutex);
return NEU_ERR_DATALAYERS_IS_NULL;
}
if (plugin->common.link_state != NEU_NODE_LINK_STATE_CONNECTED) {
pthread_mutex_unlock(&plugin->plugin_mutex);
return NEU_ERR_PLUGIN_NOT_RUNNING;
}
@@ -353,6 +358,7 @@ int handle_trans_data(neu_plugin_t * plugin,
if (NULL == route) {
plog_error(plugin, "no route for driver:%s group:%s",
trans_data->driver, trans_data->group);
pthread_mutex_unlock(&plugin->plugin_mutex);
return NEU_ERR_GROUP_NOT_SUBSCRIBE;
}
@@ -442,6 +448,8 @@ int handle_trans_data(neu_plugin_t * plugin,
task_queue_push(plugin, task);
pthread_mutex_unlock(&plugin->plugin_mutex);
return rv;
}
plugins/datalayers/datalayers_plugin.c
+1 -1
View File
@@ -33,5 +33,5 @@ const neu_plugin_module_t neu_plugin_module = {
.type = NEU_NA_TYPE_APP,
.display = true,
.single = true,
.single_name = "datalayers",
.single_name = "Datalayers",
};
plugins/datalayers/datalayers_plugin.h
+2
View File
@@ -67,6 +67,8 @@ struct neu_plugin {
task_queue_t task_queue;
pthread_t consumer_thread;
pthread_mutex_t plugin_mutex;
int (*parse_config)(neu_plugin_t *plugin, const char *setting,
datalayers_config_t *config);
};
plugins/datalayers/datalayers_plugin_intf.c
+33 -8
View File
@@ -48,6 +48,9 @@ void stop_consumer_thread(pthread_t *consumer_thread)
int datalayers_plugin_close(neu_plugin_t *plugin)
{
const char *name = neu_plugin_module.module_name;
pthread_mutex_lock(&plugin->plugin_mutex);
plog_notice(plugin, "success to free plugin:%s", name);
stop_consumer_thread(&plugin->consumer_thread);
@@ -56,6 +59,9 @@ int datalayers_plugin_close(neu_plugin_t *plugin)
client_destroy(plugin->client);
free(plugin);
pthread_mutex_unlock(&plugin->plugin_mutex);
return NEU_ERR_SUCCESS;
}
@@ -63,6 +69,8 @@ int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
{
(void) load;
pthread_mutex_init(&plugin->plugin_mutex, NULL);
task_queue_init(&plugin->task_queue);
pthread_create(&plugin->consumer_thread, NULL,
@@ -80,10 +88,18 @@ int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
int datalayers_plugin_uninit(neu_plugin_t *plugin)
{
pthread_mutex_lock(&plugin->plugin_mutex);
stop_consumer_thread(&plugin->consumer_thread);
tasks_free(&plugin->task_queue);
datalayers_config_fini(&plugin->config);
route_tbl_free(plugin->route_tbl);
plugin->route_tbl = NULL;
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_mutex_destroy(&plugin->plugin_mutex);
plog_notice(plugin, "uninitialize plugin `%s` success",
neu_plugin_module.module_name);
@@ -93,6 +109,14 @@ int datalayers_plugin_uninit(neu_plugin_t *plugin)
static int config_datalayers_client(neu_plugin_t * plugin,
const datalayers_config_t *config)
{
if (!config->host || !config->port || !config->username ||
!config->password) {
plog_error(
plugin,
"Invalid datalayers config: NULL host/port/username/password");
return NEU_ERR_DATALAYERS_INIT_FAILURE;
}
plugin->client = client_create(config->host, config->port, config->username,
config->password);
@@ -111,7 +135,6 @@ int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting)
int rv = 0;
const char * plugin_name = neu_plugin_module.module_name;
datalayers_config_t config = { 0 };
if (plugin->client != NULL) {
client_destroy(plugin->client);
}
@@ -123,18 +146,18 @@ int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting)
}
rv = config_datalayers_client(plugin, &config);
if (0 != rv) {
plog_error(plugin, "datalayers client configuration failed");
datalayers_config_fini(&config);
return rv;
}
datalayers_config_fini(&plugin->config);
memmove(&plugin->config, &config, sizeof(config));
plog_notice(plugin, "config plugin `%s` success", plugin_name);
return NEU_ERR_SUCCESS;
if (rv == 0) {
plog_notice(plugin, "config plugin `%s` success", plugin_name);
} else {
plog_error(plugin, "datalayers client configuration failed");
}
return rv;
}
int datalayers_plugin_start(neu_plugin_t *plugin)
@@ -169,6 +192,8 @@ int datalayers_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
case NEU_REQRESP_TRANS_DATA: {
if (plugin->client == NULL) {
plog_notice(plugin, "datalayers client is NULL, reconnect");
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
1, NULL);
config_datalayers_client(plugin, &plugin->config);
}
plugins/restful/adapter_handle.c
+3 -3
View File
@@ -45,7 +45,7 @@ void handle_add_adapter(nng_aio *aio)
if (strlen(req->name) >= NEU_NODE_NAME_LEN) {
CHECK_NODE_NAME_LENGTH_ERR;
} else if (strcmp(req->name, "monitor") == 0 ||
strcmp(req->name, "datalayers") == 0) {
strcmp(req->name, "Datalayers") == 0) {
CHECK_NODE_SINGLE_ERR;
} else {
int ret = 0;
@@ -89,8 +89,8 @@ static inline int send_node_update_req(nng_aio * aio,
if (strcmp(req->new_name, "monitor") == 0 ||
strcmp(req->name, "monitor") == 0 ||
strcmp(req->new_name, "datalayers") == 0 ||
strcmp(req->name, "datalayers") == 0) {
strcmp(req->new_name, "Datalayers") == 0 ||
strcmp(req->name, "Datalayers") == 0) {
return NEU_ERR_NODE_NOT_ALLOW_UPDATE;
}