From 95db2396babba59a79b27999eadfe5aaf200c6d0 Mon Sep 17 00:00:00 2001 From: fengzero Date: Tue, 14 Jun 2022 07:46:14 +0000 Subject: [PATCH 1/2] feat(msg): notify app subscribe/unsubscribe info --- include/neuron/adapter_msg.h | 20 ++++++++++++ plugins/ekuiper/plugin_ekuiper.c | 7 +++++ plugins/mqtt/mqtt.c | 7 +++++ src/adapter/adapter.c | 3 ++ src/core/manager.c | 7 +++++ src/core/manager_internal.c | 53 ++++++++++++++++++++++++++++++++ src/core/manager_internal.h | 8 +++++ 7 files changed, 105 insertions(+) diff --git a/include/neuron/adapter_msg.h b/include/neuron/adapter_msg.h index 8f91159d..c7f0c979 100644 --- a/include/neuron/adapter_msg.h +++ b/include/neuron/adapter_msg.h @@ -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; diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index be1f1a7a..cbba9817 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -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; diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 0ee94414..eb07cc96 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -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; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 94c6775c..afa84756 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -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; } diff --git a/src/core/manager.c b/src/core/manager.c index 5ebc6263..01e7d030 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -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; diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 7bf3b7c9..c218c7a3 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -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); + } +} \ No newline at end of file diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 306ff889..c67314be 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -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 From 123c5943efd07cdc7de9fd05c1a81d7da2767327 Mon Sep 17 00:00:00 2001 From: fengzero Date: Tue, 14 Jun 2022 09:38:35 +0000 Subject: [PATCH 2/2] remove tag_size in group_timer, add tag_size in plugin common --- ft/tag_test.robot | 18 ++++-------------- include/neuron/plugin.h | 16 +++++++--------- src/adapter/driver/driver.c | 14 ++++++++------ src/base/group.c | 28 +++++++++++++++------------- src/base/neu_plugin_common.c | 12 +----------- 5 files changed, 35 insertions(+), 53 deletions(-) diff --git a/ft/tag_test.robot b/ft/tag_test.robot index 66a2c815..1abd7b11 100644 --- a/ft/tag_test.robot +++ b/ft/tag_test.robot @@ -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 diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 02d63424..769491fc 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -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 diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index cb74a20e..b02fda21 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -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, diff --git a/src/base/group.c b/src/base/group.c index 3bd1aa05..4f76a2a0 100644 --- a/src/base/group.c +++ b/src/base/group.c @@ -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) diff --git a/src/base/neu_plugin_common.c b/src/base/neu_plugin_common.c index ffaf8e94..c5313e9c 100644 --- a/src/base/neu_plugin_common.c +++ b/src/base/neu_plugin_common.c @@ -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)