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

Merge pull request #537 from fengzeroz/main

Browse Source 
feat(msg): notify app subscribe/unsubscribe info
This commit is contained in:
fengzeroz
2022-06-14 17:49:53 +08:00
committed by GitHub
parent 363ec1e5f6 123c5943ef
commit 900a588c82
12 changed files with 140 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/tag_test.robot
+4 -14
View File
@@ -86,17 +86,17 @@ Delete tag from non-existent group, it should return success
Check Response Status ${res} 200
Check Error Code ${res} ${NEU_ERR_SUCCESS}
Update non-existent tag, it should return success.
Update non-existent tag, it should return failure.
${res}= Update Tags modbus-node group ${tag4}
Check Response Status ${res} 200
Check Error Code ${res} ${NEU_ERR_SUCCESS}
Check Error Code ${res} ${NEU_ERR_TAG_NOT_EXIST}
${res}= Get Tags modbus-node group
Check Response Status ${res} 200
${len} = Get Length ${res}[tags]
Should Be Equal As Integers ${len} 3
Should Be Equal As Integers ${len} 2
Should Be Equal As Strings ${res}[tags][1][name] tag1
Should Be Equal As Strings ${res}[tags][1][address] 1!400002
@@ -108,11 +108,6 @@ Update non-existent tag, it should return success.
Should Be Equal As Strings ${res}[tags][0][attribute] ${TAG_ATTRIBUTE_RW}
Should Be Equal As Strings ${res}[tags][0][type] ${TAG_TYPE_BIT}
Should Be Equal As Strings ${res}[tags][2][name] tag4
Should Be Equal As Strings ${res}[tags][2][address] 1!00031
Should Be Equal As Strings ${res}[tags][2][attribute] ${TAG_ATTRIBUTE_RW}
Should Be Equal As Strings ${res}[tags][2][type] ${TAG_TYPE_BIT}
Delete tags, it should return success
Del Tags modbus-node group "tag1","tag2"
@@ -120,9 +115,4 @@ Delete tags, it should return success
Check Response Status ${res} 200
${len} = Get Length ${res}[tags]
Should Be Equal As Integers ${len} 1
Should Be Equal As Strings ${res}[tags][0][name] tag4
Should Be Equal As Strings ${res}[tags][0][address] 1!00031
Should Be Equal As Strings ${res}[tags][0][attribute] ${TAG_ATTRIBUTE_RW}
Should Be Equal As Strings ${res}[tags][0][type] ${TAG_TYPE_BIT}
Should Be Equal As Integers ${len} 0
include/neuron/adapter_msg.h
+20
View File
@@ -55,6 +55,9 @@ typedef enum neu_reqresp_type {
NEU_REQ_GET_SUBSCRIBE_GROUP,
NEU_RESP_GET_SUBSCRIBE_GROUP,
NEU_REQ_APP_SUBSCRIBE_GROUP,
NEU_REQ_APP_UNSUBSCRIBE_GROUP,
NEU_REQ_NODE_INIT,
NEU_REQ_ADD_NODE,
NEU_REQ_DEL_NODE,
@@ -201,6 +204,17 @@ typedef struct neu_resp_get_tag {
UT_array *tags; // array neu_datatag_t
} neu_resp_get_tag_t;
typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
UT_array *tags; // array neu_datatag_t
} neu_req_app_subscribe_group_t;
typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
} neu_req_app_unsubscribe_group_t;
typedef struct {
char app[NEU_NODE_NAME_LEN];
char driver[NEU_NODE_NAME_LEN];
@@ -407,6 +421,12 @@ inline static nng_msg *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_UPDATE_LICENSE:
data_size = sizeof(neu_req_update_license_t);
break;
case NEU_REQ_APP_SUBSCRIBE_GROUP:
data_size = sizeof(neu_req_app_subscribe_group_t);
break;
case NEU_REQ_APP_UNSUBSCRIBE_GROUP:
data_size = sizeof(neu_req_app_unsubscribe_group_t);
break;
default:
assert(false);
break;
include/neuron/plugin.h
+7 -9
View File
@@ -34,12 +34,15 @@ extern "C" {
#define NEURON_PLUGIN_VER_2_0 200
typedef struct neu_plugin_common {
uint32_t magic;
neu_plugin_link_state_e link_state;
uint32_t magic;
neu_adapter_t * adapter;
const adapter_callbacks_t *adapter_callbacks;
uint32_t event_id;
zlog_category_t * log;
neu_plugin_link_state_e link_state;
uint16_t tag_size;
zlog_category_t *log;
} neu_plugin_common_t;
typedef struct neu_plugin neu_plugin_t;
@@ -50,9 +53,6 @@ struct neu_plugin_group {
char * group_name;
UT_array *tags;
// the number of all tags under the node
uint16_t tag_size;
void * user_data;
neu_plugin_group_free group_free;
};
@@ -98,8 +98,6 @@ neu_plugin_to_plugin_common(neu_plugin_t *plugin)
void neu_plugin_common_init(neu_plugin_common_t *common);
bool neu_plugin_common_check(neu_plugin_t *plugin);
uint32_t neu_plugin_get_event_id(neu_plugin_t *plugin);
int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data);
#ifdef __cplusplus
plugins/ekuiper/plugin_ekuiper.c
+7
View File
@@ -186,6 +186,13 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
send_data(plugin, trans_data);
break;
}
case NEU_REQ_APP_SUBSCRIBE_GROUP: {
neu_req_app_subscribe_group_t *sub =
(neu_req_app_subscribe_group_t *) data;
utarray_free(sub->tags);
break;
}
default:
plog_notice(plugin, "unsupported request type: %d", header->type);
break;
plugins/mqtt/mqtt.c
+7
View File
@@ -739,6 +739,13 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
case NEU_REQRESP_TRANS_DATA:
error = trans_data(plugin, data);
break;
case NEU_REQ_APP_SUBSCRIBE_GROUP: {
neu_req_app_subscribe_group_t *sub =
(neu_req_app_subscribe_group_t *) data;
utarray_free(sub->tags);
break;
}
default:
error = NEU_ERR_FAILURE;
break;
src/adapter/adapter.c
+3
View File
@@ -574,6 +574,8 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQRESP_PERSISTENCE_LOAD:
persister_singleton_load_data(adapter);
break;
case NEU_REQ_APP_SUBSCRIBE_GROUP:
case NEU_REQ_APP_UNSUBSCRIBE_GROUP:
case NEU_RESP_GET_NODE_STATE:
case NEU_RESP_GET_NODE_SETTING:
case NEU_REQRESP_TRANS_DATA:
@@ -656,6 +658,7 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
default:
assert(false);
break;
}
src/adapter/driver/driver.c
+8 -6
View File
@@ -57,7 +57,6 @@ struct neu_adapter_driver {
nng_mtx * mtx;
struct group *groups;
uint16_t tag_size;
};
static int report_callback(void *usr_data);
@@ -70,7 +69,8 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_dvalue_t value);
static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
{
neu_reqresp_head_t *req = (neu_reqresp_head_t *) r;
neu_resp_error_t nerror = { .error = error };
@@ -339,7 +339,8 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
neu_group_destroy(find->group);
free(find);
driver->tag_size -= tag_size;
neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin)
->tag_size -= tag_size;
ret = NEU_ERR_SUCCESS;
}
@@ -419,7 +420,8 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
nng_mtx_unlock(driver->mtx);
if (ret == NEU_ERR_SUCCESS) {
driver->tag_size += 1;
neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin)
->tag_size += 1;
}
return ret;
@@ -441,7 +443,8 @@ int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,
nng_mtx_unlock(driver->mtx);
if (ret == NEU_ERR_SUCCESS) {
driver->tag_size -= 1;
neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin)
->tag_size -= 1;
}
return ret;
@@ -557,7 +560,6 @@ static void group_change(void *arg, int64_t timestamp, UT_array *tags,
neu_plugin_group_t grp = {
.group_name = strdup(group->name),
.tag_size = group->driver->tag_size,
.tags = NULL,
.group_free = NULL,
.user_data = NULL,
src/base/group.c
+15 -13
View File
@@ -150,7 +150,8 @@ int neu_group_add_tag(neu_group_t *group, neu_datatag_t *tag)
int neu_group_update_tag(neu_group_t *group, neu_datatag_t *tag)
{
tag_elem_t *el = NULL;
tag_elem_t *el = NULL;
int ret = NEU_ERR_TAG_NOT_EXIST;
nng_mtx_lock(group->mtx);
@@ -162,22 +163,23 @@ int neu_group_update_tag(neu_group_t *group, neu_datatag_t *tag)
free(el->tag->addr_str);
free(el->tag);
free(el);
el = calloc(1, sizeof(tag_elem_t));
el->name = strdup(tag->name);
el->tag = calloc(1, sizeof(neu_datatag_t));
el->tag->name = strdup(tag->name);
el->tag->addr_str = strdup(tag->addr_str);
el->tag->type = tag->type;
el->tag->attribute = tag->attribute;
HASH_ADD_STR(group->tags, name, el);
update_timestamp(group);
ret = NEU_ERR_SUCCESS;
}
el = calloc(1, sizeof(tag_elem_t));
el->name = strdup(tag->name);
el->tag = calloc(1, sizeof(neu_datatag_t));
el->tag->name = strdup(tag->name);
el->tag->addr_str = strdup(tag->addr_str);
el->tag->type = tag->type;
el->tag->attribute = tag->attribute;
HASH_ADD_STR(group->tags, name, el);
update_timestamp(group);
nng_mtx_unlock(group->mtx);
return 0;
return ret;
}
int neu_group_del_tag(neu_group_t *group, const char *tag_name)
src/base/neu_plugin_common.c
+1 -11
View File
@@ -26,19 +26,9 @@
#define NEU_PLUGIN_MAGIC_NUMBER 0x43474d50 // a string "PMGC"
#define NEU_SELF_NODE_NAME_UNKNOWN "unknown-adapter"
uint32_t neu_plugin_get_event_id(neu_plugin_t *plugin)
{
neu_plugin_common_t *common = neu_plugin_to_plugin_common(plugin);
common->event_id += 2; // for avoid check event_id == 0
return common->event_id;
}
void neu_plugin_common_init(neu_plugin_common_t *common)
{
common->event_id = 1;
common->magic = NEU_PLUGIN_MAGIC_NUMBER;
common->magic = NEU_PLUGIN_MAGIC_NUMBER;
}
bool neu_plugin_common_check(neu_plugin_t *plugin)
src/core/manager.c
+7
View File
@@ -286,6 +286,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (resp.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i].addr_str);
@@ -307,6 +308,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i]);
@@ -327,6 +329,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (resp.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i].addr_str);
@@ -365,6 +368,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_sub(manager, cmd->app, cmd->driver,
cmd->group);
}
header->type = NEU_RESP_ERROR;
@@ -381,6 +386,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_unsub(manager, cmd->app, cmd->driver,
cmd->group);
}
header->type = NEU_RESP_ERROR;
src/core/manager_internal.c
+53
View File
@@ -16,6 +16,9 @@
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "adapter.h"
#include "adapter/adapter_internal.h"
#include "adapter/driver/driver_internal.h"
@@ -381,3 +384,53 @@ int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
return -1;
}
void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app,
const char *driver, const char *group)
{
nng_msg * msg;
neu_reqresp_head_t header = { 0 };
neu_req_app_subscribe_group_t sub = { 0 };
nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app);
header.type = NEU_REQ_APP_SUBSCRIBE_GROUP;
strcpy(sub.driver, driver);
strcpy(sub.group, group);
int ret = neu_manager_get_tag(manager, driver, group, &sub.tags);
assert(ret == 0);
msg = neu_msg_gen(&header, (void *) &sub);
nng_msg_set_pipe(msg, pipe);
nng_sendmsg(manager->socket, msg, 0);
}
void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app,
const char *driver, const char *group)
{
nng_msg * msg;
neu_reqresp_head_t header = { 0 };
neu_req_app_unsubscribe_group_t unsub = { 0 };
nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app);
header.type = NEU_REQ_APP_UNSUBSCRIBE_GROUP;
strcpy(unsub.driver, driver);
strcpy(unsub.group, group);
msg = neu_msg_gen(&header, (void *) &unsub);
nng_msg_set_pipe(msg, pipe);
nng_sendmsg(manager->socket, msg, 0);
}
void neu_manager_notify_app_sub_update(neu_manager_t *manager,
const char *driver, const char *group)
{
UT_array *apps =
neu_subscribe_manager_find(manager->subscribe_manager, driver, group);
if (apps != NULL) {
utarray_foreach(apps, neu_app_subscribe_t *, app)
{
neu_manager_notify_app_sub(manager, app->app_name, driver, group);
}
utarray_free(apps);
}
}
src/core/manager_internal.h
+8
View File
@@ -91,4 +91,12 @@ int neu_manager_node_get_setting(neu_manager_t *manager, const char *node,
int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
neu_persist_adapter_info_t *info);
void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
void neu_manager_notify_app_sub_update(neu_manager_t *manager,
const char *driver, const char *group);
#endif