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

Merge pull request #1449 from eeff/eeff/dev

Browse Source 
feat(restful): group update api supports renaming group
This commit is contained in:
fengzeroz
2023-08-14 11:01:08 +08:00
committed by GitHub
parent 704d194539 0d648cad5b
commit 0333ff04cf
22 changed files with 540 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/group_test.robot
+142
View File
@@ -5,6 +5,8 @@ Resource resource/error.resource
Suite Setup Start Neuronx
Suite Teardown Stop Neuronx
*** Variables ***
${g_long_str} ${{'a'*200}}
*** Test Cases ***
Add group under non-existent node, it should return failure
@@ -65,3 +67,143 @@ Add group under app node, it should return failure
Check Error Code ${res} ${NEU_ERR_GROUP_NOT_ALLOW}
[Teardown] Del Node mqtt
Update group with nonexistent node, it should fail.
${res} = Update Group modbus-node group interval=${1000}
Check Response Status ${res} 404
Check Error Code ${res} ${NEU_ERR_NODE_NOT_EXIST}
[Teardown] Del Node modbus-node
Update group using too long node name, it should fail.
${res} = Update Group ${g_long_str} group interval=${1000}
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_NODE_NAME_TOO_LONG}
[Teardown] Del Node modbus-node
Update group with nonexistent group, it should fail.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
${res} = Update Group modbus-node group interval=${1000}
Check Response Status ${res} 404
Check Error Code ${res} ${NEU_ERR_GROUP_NOT_EXIST}
[Teardown] Del Node modbus-node
Update group using to long group name, it should fail.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
${res} = Update Group modbus-node ${g_long_str} interval=${1000}
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_GROUP_NAME_TOO_LONG}
[Teardown] Del Node modbus-node
Update group with too small interval, it should fail.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
${res} = Update Group modbus-node group interval=${10}
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_GROUP_PARAMETER_INVALID}
[Teardown] Del Node modbus-node
Update group with too large interval, it should fail.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
${res} = Update Group modbus-node group interval=${100000000000}
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_GROUP_PARAMETER_INVALID}
[Teardown] Del Node modbus-node
Update group interval, it should success.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
${res} = Update Group modbus-node group interval=${2000}
Check Response Status ${res} 200
Check Error Code ${res} ${NEU_ERR_SUCCESS}
${res} = Get Group modbus-node
Set Local Variable ${groups} ${res}[groups]
Should Match Group ${groups} group ${2000}
[Teardown] Del Node modbus-node
Update group with empty new name, it should fail.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
${res} = Update Group modbus-node group new_name=${EMPTY}
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_BODY_IS_WRONG}
[Teardown] Del Node modbus-node
Update group with too long new name, it should fail.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
${res} = Update Group modbus-node group new_name=${g_long_str}
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_GROUP_NAME_TOO_LONG}
[Teardown] Del Node modbus-node
Update group name, it should success.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
${res} = Update Group modbus-node group new_name=grp
Check Response Status ${res} 200
Check Error Code ${res} ${NEU_ERR_SUCCESS}
${res} = Get Group modbus-node
Set Local Variable ${groups} ${res}[groups]
Should Match Group ${groups} grp ${1000}
[Teardown] Del Node modbus-node
Update group name and interval, it should success.
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
${res} = Update Group modbus-node group new_name=grp interval=${2000}
Check Response Status ${res} 200
Check Error Code ${res} ${NEU_ERR_SUCCESS}
${res} = Get Group modbus-node
Set Local Variable ${groups} ${res}[groups]
Should Match Group ${groups} grp ${2000}
[Teardown] Del Node modbus-node
Update group name, subscriptions should be updated.
Add Node mqtt-node ${PLUGIN-MQTT}
Add Node modbus-node ${PLUGIN-MODBUS-TCP}
Add Group modbus-node group 1000
Subscribe Group mqtt-node modbus-node group
${res} = Update Group modbus-node group new_name=grp
Check Response Status ${res} 200
Check Error Code ${res} ${NEU_ERR_SUCCESS}
${res} = Get Subscribe Group mqtt-node
Check Response Status ${res} 200
Set Local Variable ${groups} ${res}[groups]
List Length Should Be ${groups} 1
Should Be Equal As Strings ${groups}[0][driver] modbus-node
Should Be Equal As Strings ${groups}[0][group] grp
[Teardown] Del Node modbus-node
*** Keywords ***
Should Match Group
[Arguments] ${groups} ${name} ${interval}
List Length Should Be ${groups} 1
Should Be Equal As Strings ${name} ${groups}[0][name]
Should Be Equal As Integers ${interval} ${groups}[0][interval]
ft/resource/api.resource
+8
View File
@@ -117,6 +117,14 @@ Add Group
${res} = Object response body
[RETURN] ${res}[0]
Update Group
[Arguments] ${node} ${group} &{updates}
${dict} = Evaluate {"node":$node, "group":$group, **$updates}
${data} = Evaluate json.dumps(${dict}) json
PUT /api/v2/group ${data}
${res} = Object response body
[RETURN] ${res}[0]
Del Group
[Arguments] ${node} ${group}
DELETE /api/v2/group {"node": "${node}", "group": "${group}"}
include/neuron/adapter.h
+17 -1
View File
@@ -76,6 +76,7 @@ typedef enum neu_reqresp_type {
NEU_REQ_ADD_GROUP,
NEU_REQ_DEL_GROUP,
NEU_REQ_UPDATE_GROUP,
NEU_RESP_UPDATE_GROUP,
NEU_REQ_GET_GROUP,
NEU_RESP_GET_GROUP,
NEU_REQ_GET_DRIVER_GROUP,
@@ -180,6 +181,7 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_ADD_GROUP] = "NEU_REQ_ADD_GROUP",
[NEU_REQ_DEL_GROUP] = "NEU_REQ_DEL_GROUP",
[NEU_REQ_UPDATE_GROUP] = "NEU_REQ_UPDATE_GROUP",
[NEU_RESP_UPDATE_GROUP] = "NEU_RESP_UPDATE_GROUP",
[NEU_REQ_GET_GROUP] = "NEU_REQ_GET_GROUP",
[NEU_RESP_GET_GROUP] = "NEU_RESP_GET_GROUP",
[NEU_REQ_GET_DRIVER_GROUP] = "NEU_REQ_GET_DRIVER_GROUP",
@@ -328,7 +330,21 @@ typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
uint32_t interval;
} neu_req_add_group_t, neu_req_update_group_t;
} neu_req_add_group_t;
typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char new_name[NEU_GROUP_NAME_LEN];
uint32_t interval;
} neu_req_update_group_t;
typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char new_name[NEU_GROUP_NAME_LEN];
int error;
} neu_resp_update_group_t;
typedef struct neu_req_del_group {
char driver[NEU_NODE_NAME_LEN];
include/neuron/persist/persist.h
+2 -1
View File
@@ -270,8 +270,9 @@ int neu_persister_store_group(const char * driver_name,
* @param group_info group info to persist.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_update_group(const char * driver_name,
int neu_persister_update_group(const char *driver_name, const char *group_name,
neu_persist_group_info_t *group_info);
/**
* Load all group config infos under an adapter.
* @param driver_name name of the driver who owns the group
plugins/monitor/mqtt_handle.c
+14 -11
View File
@@ -65,15 +65,15 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
{
char *json_str = NULL;
union {
neu_json_add_node_req_t add_node;
neu_json_del_node_req_t del_node;
neu_json_node_ctl_req_t node_ctl;
neu_json_node_setting_req_t node_setting;
neu_json_add_group_config_req_t add_grp;
neu_json_del_group_config_req_t del_grp;
neu_json_update_group_req_t update_grp;
neu_json_add_tags_req_t add_tags;
neu_json_del_tags_req_t del_tags;
neu_json_add_node_req_t add_node;
neu_json_del_node_req_t del_node;
neu_json_node_ctl_req_t node_ctl;
neu_json_node_setting_req_t node_setting;
neu_json_add_group_config_req_t add_grp;
neu_json_del_group_config_req_t del_grp;
neu_json_update_group_config_req_t update_grp;
neu_json_add_tags_req_t add_tags;
neu_json_del_tags_req_t del_tags;
} json_req = {};
switch (event) {
@@ -135,8 +135,11 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
neu_req_update_group_t *update_grp = data;
json_req.update_grp.node = update_grp->driver;
json_req.update_grp.group = update_grp->group;
json_req.update_grp.interval = update_grp->interval;
*topic_p = plugin->config->group_update_topic;
json_req.update_grp.new_name = update_grp->new_name;
json_req.update_grp.set_interval =
update_grp->interval >= NEU_GROUP_INTERVAL_LIMIT;
json_req.update_grp.interval = update_grp->interval;
*topic_p = plugin->config->group_update_topic;
neu_json_encode_by_fn(&json_req, neu_json_encode_add_group_config_req,
&json_str);
break;
plugins/restful/group_config_handle.c
+57 -16
View File
@@ -61,27 +61,68 @@ void handle_add_group_config(nng_aio *aio)
})
}
void handle_update_group(nng_aio *aio)
static inline int send_update_group(nng_aio * aio,
neu_json_update_group_config_req_t *req)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_update_group_req_t, neu_json_decode_add_group_config_req,
{
int ret = 0;
neu_reqresp_head_t header = { 0 };
neu_req_add_group_t cmd = { 0 };
if (strlen(req->node) >= NEU_NODE_NAME_LEN) {
return NEU_ERR_NODE_NAME_TOO_LONG;
} else if (strlen(req->group) >= NEU_GROUP_NAME_LEN) {
return NEU_ERR_GROUP_NAME_TOO_LONG;
} else if (req->set_interval &&
(req->interval < NEU_GROUP_INTERVAL_LIMIT ||
req->interval > UINT32_MAX)) {
return NEU_ERR_GROUP_PARAMETER_INVALID;
}
header.ctx = aio;
header.type = NEU_REQ_UPDATE_GROUP;
strcpy(cmd.driver, req->node);
strcpy(cmd.group, req->group);
cmd.interval = req->interval;
ret = neu_plugin_op(plugin, header, &cmd);
// for backward compatibility,
// `new_name` or `interval` (inclusive) should be provided
if (!req->new_name && !req->set_interval) {
return NEU_ERR_BODY_IS_WRONG;
}
neu_req_update_group_t cmd = { 0 };
if (req->new_name) {
// if `new_name` is provided, then it should be valid
int len = strlen(req->new_name);
if (0 == len) {
return NEU_ERR_BODY_IS_WRONG;
} else if (len >= NEU_GROUP_NAME_LEN) {
return NEU_ERR_GROUP_NAME_TOO_LONG;
}
strcpy(cmd.new_name, req->new_name);
} else {
// if `new_name` is omitted, then keep the node name
strcpy(cmd.new_name, req->group);
}
strcpy(cmd.driver, req->node);
strcpy(cmd.group, req->group);
cmd.interval = req->set_interval ? req->interval : 0;
neu_reqresp_head_t header = {
.ctx = aio,
.type = NEU_REQ_UPDATE_GROUP,
};
if (0 != neu_plugin_op(plugin, header, &cmd)) {
return NEU_ERR_IS_BUSY;
}
return 0;
}
void handle_update_group(nng_aio *aio)
{
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_update_group_config_req_t,
neu_json_decode_update_group_config_req, {
int ret = send_update_group(aio, req);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
NEU_JSON_RESPONSE_ERROR(
ret, { neu_http_response(aio, ret, result_error); });
}
})
}
src/adapter/adapter.c
+45 -15
View File
@@ -665,30 +665,29 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_REQ_UPDATE_GROUP: {
neu_req_update_group_t *cmd = (neu_req_update_group_t *) &header[1];
neu_resp_error_t error = { 0 };
neu_req_update_group_t *cmd = (neu_req_update_group_t *) &header[1];
neu_resp_update_group_t resp = { 0 };
if (cmd->interval < NEU_GROUP_INTERVAL_LIMIT) {
error.error = NEU_ERR_GROUP_PARAMETER_INVALID;
if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
resp.error = neu_adapter_driver_update_group(
(neu_adapter_driver_t *) adapter, cmd->group, cmd->new_name,
cmd->interval);
} else {
if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
error.error = neu_adapter_driver_update_group(
(neu_adapter_driver_t *) adapter, cmd->group,
cmd->interval);
} else {
error.error = NEU_ERR_GROUP_NOT_ALLOW;
}
resp.error = NEU_ERR_GROUP_NOT_ALLOW;
}
if (error.error == NEU_ERR_SUCCESS) {
if (resp.error == NEU_ERR_SUCCESS) {
adapter_storage_update_group(adapter->name, cmd->group,
cmd->interval);
cmd->new_name, cmd->interval);
notify_monitor(adapter, NEU_REQ_UPDATE_GROUP_EVENT, cmd);
}
strcpy(resp.driver, cmd->driver);
strcpy(resp.group, cmd->group);
strcpy(resp.new_name, cmd->new_name);
header->type = NEU_RESP_UPDATE_GROUP;
neu_msg_exchange(header);
header->type = NEU_RESP_ERROR;
reply(adapter, header, &error);
reply(adapter, header, &resp);
break;
}
case NEU_REQ_DEL_GROUP: {
@@ -1213,6 +1212,34 @@ int neu_adapter_update_group_metric(neu_adapter_t *adapter,
return 0;
}
int neu_adapter_metric_update_group_name(neu_adapter_t *adapter,
const char * group_name,
const char * new_group_name)
{
neu_group_metrics_t *group_metrics = NULL;
if (NULL == adapter->metrics) {
return -1;
}
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
if (NULL == group_metrics) {
return -1;
}
char *name = strdup(new_group_name);
if (NULL == name) {
return -1;
}
HASH_DEL(adapter->metrics->group_metrics, group_metrics);
free(group_metrics->name);
group_metrics->name = name;
HASH_ADD_STR(adapter->metrics->group_metrics, name, group_metrics);
return 0;
}
void neu_adapter_del_group_metrics(neu_adapter_t *adapter,
const char * group_name)
{
@@ -1384,6 +1411,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_UPDATE_GROUP_EVENT:
data_size = sizeof(neu_req_update_group_t);
break;
case NEU_RESP_UPDATE_GROUP:
data_size = sizeof(neu_resp_update_group_t);
break;
case NEU_REQ_DEL_GROUP:
case NEU_REQ_DEL_GROUP_EVENT:
data_size = sizeof(neu_req_del_group_t);
src/adapter/adapter_internal.h
+3
View File
@@ -94,6 +94,9 @@ int neu_adapter_register_group_metric(neu_adapter_t *adapter,
int neu_adapter_update_group_metric(neu_adapter_t *adapter,
const char * group_name,
const char *metric_name, uint64_t n);
int neu_adapter_metric_update_group_name(neu_adapter_t *adapter,
const char * group_name,
const char * new_group_name);
void neu_adapter_del_group_metrics(neu_adapter_t *adapter,
const char * group_name);
src/adapter/driver/driver.c
+55 -22
View File
@@ -712,34 +712,67 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
}
int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
const char *name, uint32_t interval)
const char *name, const char *new_name,
uint32_t interval)
{
int ret = 0;
group_t *find = NULL;
int ret = NEU_ERR_GROUP_NOT_EXIST;
HASH_FIND_STR(driver->groups, name, find);
if (find != NULL) {
neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
neu_event_del_timer(driver->driver_events, find->read);
neu_event_timer_param_t param = {
.second = interval / 1000,
.millisecond = interval % 1000,
.usr_data = (void *) find,
.type = NEU_EVENT_TIMER_NOBLOCK,
};
neu_group_set_interval(find->group, interval);
param.cb = report_callback;
find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
param.type = driver->adapter.module->timer_type;
param.cb = read_callback;
find->read = neu_event_add_timer(driver->driver_events, param);
ret = NEU_ERR_SUCCESS;
if (NULL == find) {
return NEU_ERR_GROUP_NOT_EXIST;
}
if (NULL != new_name && 0 == strlen(new_name)) {
return NEU_ERR_EINTERNAL;
}
// stop the timer first to avoid race condition
neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
neu_event_del_timer(driver->driver_events, find->read);
// a diminutive value should keep the interval untouched
if (interval < NEU_GROUP_INTERVAL_LIMIT) {
interval = neu_group_get_interval(find->group);
}
// update group name if a different name is provided
if (NULL != new_name && 0 != strcmp(name, new_name)) {
char *new_name_cp1 = strdup(new_name);
char *new_name_cp2 = strdup(new_name);
if (new_name_cp1 && new_name_cp2 &&
0 == neu_group_set_name(find->group, new_name)) {
HASH_DEL(driver->groups, find);
free(find->name);
find->name = new_name_cp1;
free(find->grp.group_name);
find->grp.group_name = new_name_cp2;
neu_adapter_metric_update_group_name((neu_adapter_t *) driver, name,
new_name);
HASH_ADD_STR(driver->groups, name, find);
} else {
free(new_name_cp1);
free(new_name_cp2);
ret = NEU_ERR_EINTERNAL;
}
}
neu_event_timer_param_t param = {
.second = interval / 1000,
.millisecond = interval % 1000,
.usr_data = (void *) find,
.type = NEU_EVENT_TIMER_NOBLOCK,
};
neu_group_set_interval(find->group, interval);
// restore the timers
param.cb = report_callback;
find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
param.type = driver->adapter.module->timer_type;
param.cb = read_callback;
find->read = neu_event_add_timer(driver->driver_events, param);
return ret;
}
src/adapter/driver/driver_internal.h
+2 -1
View File
@@ -44,7 +44,8 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
uint32_t interval);
int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
const char *name, uint32_t interval);
const char *name, const char *new_name,
uint32_t interval);
int neu_adapter_driver_del_group(neu_adapter_driver_t *driver,
const char * name);
int neu_adapter_driver_group_exist(neu_adapter_driver_t *driver,
src/adapter/storage.c
+3 -3
View File
@@ -45,14 +45,14 @@ void adapter_storage_add_group(const char *node, const char *group,
}
void adapter_storage_update_group(const char *node, const char *group,
uint32_t interval)
const char *new_name, uint32_t interval)
{
neu_persist_group_info_t info = {
.name = (char *) group,
.name = (char *) new_name,
.interval = interval,
};
int rv = neu_persister_update_group(node, &info);
int rv = neu_persister_update_group(node, group, &info);
if (0 != rv) {
nlog_error("fail update adapter:%s group:%s, interval: %u", node, group,
interval);
src/adapter/storage.h
+1 -1
View File
@@ -29,7 +29,7 @@ void adapter_storage_setting(const char *node, const char *setting);
void adapter_storage_add_group(const char *node, const char *group,
uint32_t interval);
void adapter_storage_update_group(const char *node, const char *group,
uint32_t interval);
const char *new_name, uint32_t interval);
void adapter_storage_del_group(const char *node, const char *group);
void adapter_storage_add_tag(const char *node, const char *group,
const neu_datatag_t *tag);
src/base/group.c
+12
View File
@@ -78,6 +78,18 @@ const char *neu_group_get_name(const neu_group_t *group)
return group->name;
}
int neu_group_set_name(neu_group_t *group, const char *name)
{
char *new_name = NULL;
if (NULL == name || NULL == (new_name = strdup(name))) {
return NEU_ERR_EINTERNAL;
}
free(group->name);
group->name = new_name;
return 0;
}
uint32_t neu_group_get_interval(const neu_group_t *group)
{
uint32_t interval = 0;
src/base/group.h
+1
View File
@@ -30,6 +30,7 @@ typedef struct neu_group neu_group_t;
neu_group_t *neu_group_new(const char *name, uint32_t interval);
const char * neu_group_get_name(const neu_group_t *group);
int neu_group_set_name(neu_group_t *group, const char *name);
uint32_t neu_group_get_interval(const neu_group_t *group);
void neu_group_set_interval(neu_group_t *group, uint32_t interval);
void neu_group_destroy(neu_group_t *group);
src/core/manager.c
+13
View File
@@ -873,6 +873,19 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_RESP_UPDATE_GROUP: {
neu_resp_update_group_t *resp = (neu_resp_update_group_t *) &header[1];
if (0 == resp->error) {
neu_manager_update_group_name(manager, resp->driver, resp->group,
resp->new_name);
}
neu_resp_error_t e = { .error = resp->error };
header->type = NEU_RESP_ERROR;
reply(manager, header, &e);
break;
}
case NEU_REQ_ADD_NODE_EVENT:
case NEU_REQ_DEL_NODE_EVENT:
case NEU_REQ_NODE_CTL_EVENT:
src/core/manager_internal.c
+7
View File
@@ -124,6 +124,13 @@ int neu_manager_update_node_name(neu_manager_t *manager, const char *node,
return ret;
}
int neu_manager_update_group_name(neu_manager_t *manager, const char *driver,
const char *group, const char *new_name)
{
return neu_subscribe_manager_update_group_name(manager->subscribe_manager,
driver, group, new_name);
}
static inline neu_plugin_instance_t *
new_plugin_instance(neu_plugin_manager_t *plugin_manager, const char *plugin)
{
src/core/manager_internal.h
+2
View File
@@ -59,6 +59,8 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, int type,
const char *plugin, const char *node);
int neu_manager_update_node_name(neu_manager_t *manager, const char *node,
const char *new_name);
int neu_manager_update_group_name(neu_manager_t *manager, const char *driver,
const char *group, const char *new_name);
int neu_manager_add_template(neu_manager_t *manager, const char *name,
const char *plugin, uint16_t n_group,
src/core/subscribe.c
+24
View File
@@ -329,3 +329,27 @@ int neu_subscribe_manager_update_driver_name(neu_subscribe_mgr_t *mgr,
return 0;
}
int neu_subscribe_manager_update_group_name(neu_subscribe_mgr_t *mgr,
const char * driver,
const char * group,
const char * new_name)
{
sub_elem_t * find = NULL;
sub_elem_key_t key = { 0 };
strncpy(key.driver, driver, sizeof(key.driver));
strncpy(key.group, group, sizeof(key.group));
HASH_FIND(hh, mgr->ss, &key, sizeof(sub_elem_key_t), find);
if (NULL == find) {
return NEU_ERR_GROUP_NOT_SUBSCRIBE;
}
HASH_DEL(mgr->ss, find);
strcpy(find->key.group, new_name);
HASH_ADD(hh, mgr->ss, key, sizeof(sub_elem_key_t), find);
return 0;
}
src/core/subscribe.h
+4
View File
@@ -64,6 +64,10 @@ int neu_subscribe_manager_update_app_name(neu_subscribe_mgr_t *mgr,
int neu_subscribe_manager_update_driver_name(neu_subscribe_mgr_t *mgr,
const char * driver,
const char * new_name);
int neu_subscribe_manager_update_group_name(neu_subscribe_mgr_t *mgr,
const char * driver,
const char * group,
const char * new_name);
int neu_subscribe_manager_update_params(neu_subscribe_mgr_t *mgr,
const char *app, const char *driver,
const char *group, const char *params);
src/parser/neu_json_group_config.c
+90 -37
View File
@@ -540,60 +540,113 @@ void neu_json_decode_unsubscribe_req_free(neu_json_unsubscribe_req_t *req)
free(req);
}
int neu_json_encode_update_group_config_req(void *json_object, void *param)
{
int ret = 0;
neu_json_update_group_config_req_t *req = param;
neu_json_elem_t req_elems[] = { {
.name = "node",
.t = NEU_JSON_STR,
.v.val_str = req->node,
},
{
.name = "group",
.t = NEU_JSON_STR,
.v.val_str = req->group,
},
{
.name = "new_name",
.t = NEU_JSON_STR,
.v.val_str = req->new_name,
} };
ret = neu_json_encode_field(json_object, req_elems,
NEU_JSON_ELEM_SIZE(req_elems));
if (0 != ret) {
return ret;
}
if (req->set_interval) {
neu_json_elem_t interval_elem = {
.name = "interval",
.t = NEU_JSON_INT,
.v.val_int = req->interval,
};
ret = neu_json_encode_field(json_object, &interval_elem, 1);
}
return ret;
}
int neu_json_decode_update_group_config_req(
char *buf, neu_json_update_group_config_req_t **result)
{
int ret = 0;
void * json_obj = NULL;
neu_json_update_group_config_req_t *req =
calloc(1, sizeof(neu_json_update_group_config_req_t));
int ret = 0;
neu_json_update_group_config_req_t *req = calloc(1, sizeof(*req));
if (req == NULL) {
return -1;
}
json_obj = neu_json_decode_new(buf);
neu_json_elem_t req_elems[] = { {
.name = "node_name",
.t = NEU_JSON_STR,
},
{
.name = "name",
.t = NEU_JSON_STR,
},
{
.name = "interval",
.t = NEU_JSON_INT,
} };
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
if (ret != 0) {
goto decode_fail;
void *json_obj = neu_json_decode_new(buf);
if (NULL == json_obj) {
free(req);
return -1;
}
req->node_name = req_elems[0].v.val_str;
req->name = req_elems[1].v.val_str;
req->interval = req_elems[2].v.val_int;
neu_json_elem_t req_elems[] = {
{
.name = "node",
.t = NEU_JSON_STR,
},
{
.name = "group",
.t = NEU_JSON_STR,
},
{
.name = "new_name",
.t = NEU_JSON_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
};
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
req->node = req_elems[0].v.val_str;
req->group = req_elems[1].v.val_str;
req->new_name = req_elems[2].v.val_str;
if (0 != ret) {
goto error;
}
neu_json_elem_t interval_elem = {
.name = "interval",
.t = NEU_JSON_INT,
};
ret = neu_json_decode_by_json(json_obj, 1, &interval_elem);
if (0 == ret) {
req->set_interval = true;
req->interval = interval_elem.v.val_int;
} else if (NULL == req->new_name) {
// at least one of `new_name` or `interval` should be provided
goto error;
}
*result = req;
goto decode_exit;
neu_json_decode_free(json_obj);
return 0;
decode_fail:
free(req);
ret = -1;
decode_exit:
if (json_obj != NULL) {
neu_json_decode_free(json_obj);
}
return ret;
error:
neu_json_decode_update_group_config_req_free(req);
neu_json_decode_free(json_obj);
return -1;
}
void neu_json_decode_update_group_config_req_free(
neu_json_update_group_config_req_t *req)
{
free(req->node_name);
free(req->name);
free(req->node);
free(req->group);
free(req->new_name);
free(req);
}
src/parser/neu_json_group_config.h
+14 -11
View File
@@ -43,6 +43,20 @@ int neu_json_decode_add_group_config_req(
void neu_json_decode_add_group_config_req_free(
neu_json_add_group_config_req_t *req);
typedef struct {
char * node;
char * group;
char * new_name;
bool set_interval;
int64_t interval;
} neu_json_update_group_config_req_t;
int neu_json_encode_update_group_config_req(void *json_object, void *param);
int neu_json_decode_update_group_config_req(
char *buf, neu_json_update_group_config_req_t **result);
void neu_json_decode_update_group_config_req_free(
neu_json_update_group_config_req_t *req);
typedef struct {
char *group;
char *node;
@@ -120,17 +134,6 @@ int neu_json_decode_unsubscribe_req(char * buf,
neu_json_unsubscribe_req_t **result);
void neu_json_decode_unsubscribe_req_free(neu_json_unsubscribe_req_t *req);
typedef struct {
char * name;
char * node_name;
int64_t interval;
} neu_json_update_group_config_req_t;
int neu_json_decode_update_group_config_req(
char *buf, neu_json_update_group_config_req_t **result);
void neu_json_decode_update_group_config_req_free(
neu_json_update_group_config_req_t *req);
typedef struct {
char *driver;
char *group;
src/persist/persist.c
+24 -6
View File
@@ -1168,14 +1168,32 @@ int neu_persister_store_group(const char * driver_name,
driver_name, group_info->name, (unsigned) group_info->interval);
}
int neu_persister_update_group(const char * driver_name,
int neu_persister_update_group(const char *driver_name, const char *group_name,
neu_persist_group_info_t *group_info)
{
return execute_sql(global_db,
"UPDATE groups SET driver_name=%Q, name=%Q, "
"interval=%i WHERE driver_name=%Q AND name=%Q",
driver_name, group_info->name, group_info->interval,
driver_name, group_info->name);
int ret = -1;
bool update_name = (0 != strcmp(group_name, group_info->name));
bool update_interval = (NEU_GROUP_INTERVAL_LIMIT <= group_info->interval);
if (update_name && update_interval) {
ret = execute_sql(global_db,
"UPDATE groups SET name=%Q, interval=%i "
"WHERE driver_name=%Q AND name=%Q",
group_info->name, group_info->interval, driver_name,
group_name);
} else if (update_name) {
ret = execute_sql(global_db,
"UPDATE groups SET name=%Q "
"WHERE driver_name=%Q AND name=%Q",
group_info->name, driver_name, group_name);
} else if (update_interval) {
ret = execute_sql(global_db,
"UPDATE groups SET interval=%i "
"WHERE driver_name=%Q AND name=%Q",
group_info->interval, driver_name, group_name);
}
return ret;
}
static UT_icd group_info_icd = {