From a999c9f80c93331aa864a3d27f1b9454dbd2c90d Mon Sep 17 00:00:00 2001 From: hxy7yx <1595670487@qq.com> Date: Wed, 7 May 2025 01:53:24 -0700 Subject: [PATCH] fix(datalayers):restore connection --- plugins/datalayers/datalayers_handle.c | 8 ++++ plugins/datalayers/datalayers_plugin.c | 2 +- plugins/datalayers/datalayers_plugin.h | 2 + plugins/datalayers/datalayers_plugin_intf.c | 41 +++++++++++++++++---- plugins/restful/adapter_handle.c | 6 +-- 5 files changed, 47 insertions(+), 12 deletions(-) diff --git a/plugins/datalayers/datalayers_handle.c b/plugins/datalayers/datalayers_handle.c index a76368e0..7319e3a9 100644 --- a/plugins/datalayers/datalayers_handle.c +++ b/plugins/datalayers/datalayers_handle.c @@ -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; } diff --git a/plugins/datalayers/datalayers_plugin.c b/plugins/datalayers/datalayers_plugin.c index d18741a9..72465b7e 100644 --- a/plugins/datalayers/datalayers_plugin.c +++ b/plugins/datalayers/datalayers_plugin.c @@ -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", }; \ No newline at end of file diff --git a/plugins/datalayers/datalayers_plugin.h b/plugins/datalayers/datalayers_plugin.h index c20294ad..6cf53908 100644 --- a/plugins/datalayers/datalayers_plugin.h +++ b/plugins/datalayers/datalayers_plugin.h @@ -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); }; diff --git a/plugins/datalayers/datalayers_plugin_intf.c b/plugins/datalayers/datalayers_plugin_intf.c index ba94ba54..e82dea85 100644 --- a/plugins/datalayers/datalayers_plugin_intf.c +++ b/plugins/datalayers/datalayers_plugin_intf.c @@ -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); } diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index 10d11808..262f32d0 100644 --- a/plugins/restful/adapter_handle.c +++ b/plugins/restful/adapter_handle.c @@ -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; }