GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-19 14:23:05 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
v2.12
neuron/plugins/datalayers/datalayers_plugin_intf.c
T
hxy7yx 9371e834e8 fix(datalayers):queue
2025-06-26 19:04:20 -07:00

277 lines
8.7 KiB
C
Raw Permalink Blame History

/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "define.h"
#include "errcodes.h"
#include "flight_sql_client.h"
#include "plugin.h"
#include "datalayers_config.h"
#include "datalayers_handle.h"
#include "datalayers_plugin.h"
#include "datalayers_plugin_intf.h"
#include <netinet/in.h>
extern const neu_plugin_module_t neu_plugin_module;
void create_consumer_thread(neu_plugin_t *plugin)
{
pthread_mutex_lock(&plugin->queue_mutex);
plugin->consumer_thread_stop_flag = false;
pthread_mutex_unlock(&plugin->queue_mutex);
int status = pthread_create(&plugin->consumer_thread, NULL,
(void *) db_write_task_consumer, plugin);
if (status != 0) {
plog_notice(plugin, "Failed to create consumer thread: %d\n", status);
}
}
neu_plugin_t *datalayers_plugin_open(void)
{
neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
neu_plugin_common_init(&plugin->common);
plugin->parse_config = datalayers_config_parse;
return plugin;
}
void stop_consumer_thread(pthread_t *consumer_thread, neu_plugin_t *plugin)
{
pthread_mutex_lock(&plugin->queue_mutex);
plugin->consumer_thread_stop_flag = true;
pthread_cond_broadcast(&plugin->queue_not_empty);
pthread_mutex_unlock(&plugin->queue_mutex);
int join_status = pthread_join(*consumer_thread, NULL);
if (join_status != 0) {
plog_notice(plugin, "Failed to join consumer thread: %d\n",
join_status);
}
}
int datalayers_plugin_close(neu_plugin_t *plugin)
{
const char *name = neu_plugin_module.module_name;
plog_notice(plugin, "success to free plugin:%s", name);
free(plugin);
return NEU_ERR_SUCCESS;
}
int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
{
(void) load;
pthread_rwlock_init(&plugin->plugin_mutex, NULL);
pthread_mutex_init(&plugin->queue_mutex, NULL);
pthread_cond_init(&plugin->queue_not_empty, NULL);
pthread_mutex_lock(&plugin->queue_mutex);
plugin->consumer_thread_stop_flag = false;
pthread_mutex_unlock(&plugin->queue_mutex);
task_queue_init(&plugin->task_queue);
pthread_create(&plugin->consumer_thread, NULL,
(void *) db_write_task_consumer, plugin);
NEU_PLUGIN_REGISTER_CACHED_QUEUE_SIZE_METRIC(plugin);
NEU_PLUGIN_REGISTER_MAX_CACHED_QUEUE_SIZE_METRIC(plugin);
NEU_PLUGIN_REGISTER_DISCARDED_MSGS_METRIC(plugin);
plog_notice(plugin, "initialize plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
int datalayers_plugin_uninit(neu_plugin_t *plugin)
{
pthread_rwlock_wrlock(&plugin->plugin_mutex);
stop_consumer_thread(&plugin->consumer_thread, plugin);
pthread_mutex_lock(&plugin->queue_mutex);
tasks_free(&plugin->task_queue);
pthread_mutex_unlock(&plugin->queue_mutex);
pthread_mutex_destroy(&plugin->queue_mutex);
pthread_cond_destroy(&plugin->queue_not_empty);
datalayers_config_fini(&plugin->config);
route_tbl_free(plugin->route_tbl);
plugin->route_tbl = NULL;
client_destroy(plugin->client);
plugin->client = NULL;
pthread_rwlock_unlock(&plugin->plugin_mutex);
pthread_rwlock_destroy(&plugin->plugin_mutex);
plog_notice(plugin, "uninitialize plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
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);
if (NULL == plugin->client) {
plog_error(plugin, "datalayers client_create failed");
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
} else {
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
}
return NEU_ERR_SUCCESS;
}
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) {
stop_consumer_thread(&plugin->consumer_thread, plugin);
pthread_mutex_lock(&plugin->queue_mutex);
tasks_free(&plugin->task_queue);
pthread_mutex_unlock(&plugin->queue_mutex);
client_destroy(plugin->client);
plugin->client = NULL;
create_consumer_thread(plugin);
}
rv = plugin->parse_config(plugin, setting, &config);
if (0 != rv) {
plog_error(plugin, "neu_datalayers_config_parse fail");
return NEU_ERR_NODE_SETTING_INVALID;
}
rv = config_datalayers_client(plugin, &config);
datalayers_config_fini(&plugin->config);
memmove(&plugin->config, &config, sizeof(config));
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)
{
const char *plugin_name = neu_plugin_module.module_name;
if (NULL == plugin->client) {
plog_error(plugin, "datalayers started failed, reconnect");
int ret = config_datalayers_client(plugin, &plugin->config);
if (ret != 0) {
plog_error(plugin, "datalayers started failed");
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
return NEU_ERR_SUCCESS;
}
}
plog_notice(plugin, "start plugin `%s` success", plugin_name);
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
return NEU_ERR_SUCCESS;
}
int datalayers_plugin_stop(neu_plugin_t *plugin)
{
plog_notice(plugin, "stop plugin `%s` success",
neu_plugin_module.module_name);
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
return NEU_ERR_SUCCESS;
}
int datalayers_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data)
{
neu_err_code_e error = NEU_ERR_SUCCESS;
switch (head->type) {
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);
}
error = handle_trans_data(plugin, data);
break;
}
case NEU_REQ_SUBSCRIBE_GROUP:
error = handle_subscribe_group(plugin, data);
break;
case NEU_REQ_UNSUBSCRIBE_GROUP:
error = handle_unsubscribe_group(plugin, data);
break;
case NEU_REQ_UPDATE_GROUP:
error = handle_update_group(plugin, data);
break;
case NEU_REQ_DEL_GROUP:
error = handle_del_group(plugin, data);
break;
case NEU_REQ_UPDATE_NODE:
error = handle_update_driver(plugin, data);
break;
case NEU_REQRESP_NODE_DELETED: {
neu_reqresp_node_deleted_t *req = data;
if (0 != strcmp(plugin->common.name, req->node)) {
error = handle_del_driver(plugin, data);
}
break;
}
default:
error = NEU_ERR_EINTERNAL;
break;
}
return error;
}
const neu_plugin_intf_funs_t datalayers_plugin_intf_funs = {
.open = datalayers_plugin_open,
.close = datalayers_plugin_close,
.init = datalayers_plugin_init,
.uninit = datalayers_plugin_uninit,
.start = datalayers_plugin_start,
.stop = datalayers_plugin_stop,
.setting = datalayers_plugin_config,
.request = datalayers_plugin_request,
};
Reference in New Issue View Git Blame Copy Permalink