diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index ce84a01b..716546a7 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -72,7 +72,7 @@ jobs: mosquitto -v & python3 -m pip install -U pip python3 -m pip install -r ft/requirements.txt - python3 -m robot -P ft/ -d ft/reports ft + python3 -m robot --maxerrorlines=600 -P ft/ -d ft/reports ft - uses: actions/upload-artifact@v3 with: diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 71edb913..c19daeb9 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -52,6 +52,7 @@ typedef enum neu_reqresp_type { NEU_RESP_GET_SUBSCRIBE_GROUP, NEU_REQ_APP_SUBSCRIBE_GROUP, + NEU_RESP_APP_SUBSCRIBE_GROUP, NEU_REQ_APP_UNSUBSCRIBE_GROUP, NEU_REQ_NODE_INIT, @@ -206,12 +207,12 @@ 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; +} neu_resp_app_subscribe_group_t; typedef struct { char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; -} neu_req_app_unsubscribe_group_t; +} neu_req_app_subscribe_group_t, neu_req_app_unsubscribe_group_t; typedef struct { char app[NEU_NODE_NAME_LEN]; @@ -419,9 +420,10 @@ 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); + case NEU_RESP_APP_SUBSCRIBE_GROUP: + data_size = sizeof(neu_resp_app_subscribe_group_t); break; + case NEU_REQ_APP_SUBSCRIBE_GROUP: case NEU_REQ_APP_UNSUBSCRIBE_GROUP: data_size = sizeof(neu_req_app_unsubscribe_group_t); break; diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 9720f30a..ec19716d 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -170,9 +170,9 @@ 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; + case NEU_RESP_APP_SUBSCRIBE_GROUP: { + neu_resp_app_subscribe_group_t *sub = + (neu_resp_app_subscribe_group_t *) data; utarray_free(sub->tags); break; diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index e5935224..eb9c6765 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -518,11 +518,11 @@ 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; + case NEU_RESP_APP_SUBSCRIBE_GROUP: { + neu_resp_app_subscribe_group_t *sub = + (neu_resp_app_subscribe_group_t *) data; - // utarray_free(sub->tags); + utarray_free(sub->tags); break; } default: diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 7a9953df..f810526d 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -226,7 +226,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) header = (neu_reqresp_head_t *) nng_msg_body(msg); switch (header->type) { - case NEU_REQ_APP_SUBSCRIBE_GROUP: case NEU_REQ_APP_UNSUBSCRIBE_GROUP: case NEU_RESP_GET_NODE_STATE: case NEU_RESP_GET_NODE_SETTING: @@ -240,6 +239,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_GROUP: case NEU_RESP_ERROR: case NEU_REQRESP_TRANS_DATA: + case NEU_RESP_APP_SUBSCRIBE_GROUP: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); break; @@ -267,8 +267,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_msg_exchange(header); reply(adapter, header, &error); } else { + void *msg_dump = calloc(nng_msg_len(msg), 1); + memcpy(msg_dump, header, nng_msg_len(msg)); neu_adapter_driver_write_tag((neu_adapter_driver_t *) adapter, - header); + msg_dump); } break; @@ -332,6 +334,21 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } break; } + case NEU_REQ_APP_SUBSCRIBE_GROUP: { + neu_req_app_subscribe_group_t *cmd = + (neu_req_app_subscribe_group_t *) &header[1]; + neu_resp_app_subscribe_group_t resp = { 0 }; + + neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, cmd->group, + &resp.tags); + neu_msg_exchange(header); + strcpy(resp.driver, cmd->driver); + strcpy(resp.group, cmd->group); + + header->type = NEU_RESP_APP_SUBSCRIBE_GROUP; + reply(adapter, header, &resp); + break; + } case NEU_REQ_GET_TAG: { neu_req_get_tag_t *cmd = (neu_req_get_tag_t *) &header[1]; neu_resp_error_t error = { .error = 0 }; diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 6eb4109f..09e92b6e 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -76,6 +76,7 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error) req->type = NEU_RESP_ERROR; adapter->cb_funs.response(adapter, req, &nerror); + free(req); } static void update(neu_adapter_t *adapter, const char *group, const char *tag, diff --git a/src/core/manager.c b/src/core/manager.c index be3c7aa5..5beef73e 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -359,6 +359,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) reply(manager, header, &e); } else { forward_msg(manager, msg, header->receiver); + neu_manager_notify_app_unsub_update(manager, cmd->driver, + cmd->group); neu_subscribe_manager_remove(manager->subscribe_manager, cmd->driver, cmd->group); } @@ -379,6 +381,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) free(cmd->tags); } else { forward_msg(manager, msg, header->receiver); + neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group); } break; @@ -401,6 +404,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) free(cmd->tags); } else { forward_msg(manager, msg, header->receiver); + neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group); } break; @@ -429,6 +433,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_NODE_STATE: case NEU_RESP_ERROR: case NEU_RESP_READ_GROUP: + case NEU_RESP_APP_SUBSCRIBE_GROUP: forward_msg(manager, msg, header->receiver); break; default: diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 18bf246c..15ef3c7d 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -155,13 +155,14 @@ void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app, 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); + nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, driver); header.type = NEU_REQ_APP_SUBSCRIBE_GROUP; + strcpy(header.sender, app); + strcpy(header.receiver, driver); + 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); @@ -177,6 +178,9 @@ void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app, nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app); header.type = NEU_REQ_APP_UNSUBSCRIBE_GROUP; + strcpy(header.sender, "manager"); + strcpy(header.receiver, app); + strcpy(unsub.driver, driver); strcpy(unsub.group, group); @@ -185,6 +189,34 @@ void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app, nng_sendmsg(manager->socket, msg, 0); } +void neu_manager_notify_app_unsub_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_unsub(manager, app->app_name, driver, group); + } + utarray_free(apps); + } +} + +void neu_manager_notify_app_unsub_driver_update(neu_manager_t *manager, + const char * driver) +{ + UT_array *apps = neu_subscribe_manager_find_by_driver( + manager->subscribe_manager, driver); + if (apps != NULL) { + utarray_foreach(apps, neu_app_subscribe_t *, app) + { + neu_manager_notify_app_unsub(manager, app->app_name, driver, ""); + } + utarray_free(apps); + } +} + void neu_manager_notify_app_sub_update(neu_manager_t *manager, const char *driver, const char *group) { diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index b8c31a4c..ba5ba173 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -64,6 +64,10 @@ 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_unsub_update(neu_manager_t *manager, + const char *driver, const char *group); +void neu_manager_notify_app_unsub_driver_update(neu_manager_t *manager, + const char * driver); void neu_manager_notify_app_sub_update(neu_manager_t *manager, const char *driver, const char *group); diff --git a/src/core/subscribe.c b/src/core/subscribe.c index a3368b46..c9c68a4d 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -91,6 +91,24 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr, } } +UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr, + const char * driver) +{ + sub_elem_t *el = NULL, *tmp = NULL; + UT_array * apps = NULL; + + utarray_new(apps, &app_sub_icd); + + HASH_ITER(hh, mgr->ss, el, tmp) + { + if (strcmp(driver, el->key.driver) == 0) { + utarray_concat(apps, el->apps); + } + } + + return apps; +} + UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app) { sub_elem_t *el = NULL, *tmp = NULL; @@ -98,7 +116,6 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app) UT_icd icd = { sizeof(neu_resp_subscribe_info_t), NULL, NULL, NULL }; utarray_new(groups, &icd); - HASH_ITER(hh, mgr->ss, el, tmp) { utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app) diff --git a/src/core/subscribe.h b/src/core/subscribe.h index 81957f53..04906ed1 100644 --- a/src/core/subscribe.h +++ b/src/core/subscribe.h @@ -28,7 +28,7 @@ typedef struct neu_subscribe_mgr neu_subscribe_mgr_t; typedef struct neu_app_subscribe { - char app_name[128]; + char app_name[NEU_NODE_NAME_LEN]; nng_pipe pipe; } neu_app_subscribe_t; @@ -38,6 +38,8 @@ void neu_subscribe_manager_destroy(neu_subscribe_mgr_t *mgr); // neu_app_subscribe_t array UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr, const char *driver, const char *group); +UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr, + const char * driver); UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app); int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group,