Merge pull request #1320 from eeff/eeff/dev

feat(core): forward add/del ndriver map message to ndriver node

include/neuron/errcodes.h

@@ -58,6 +58,7 @@ typedef enum {
     NEU_ERR_NODE_NOT_ALLOW_DELETE    = 2011,
     NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE = 2012,
     NEU_ERR_NODE_NOT_ALLOW_UPDATE    = 2013,
+    NEU_ERR_NODE_NOT_ALLOW_MAP       = 2014,
 
     NEU_ERR_GROUP_ALREADY_SUBSCRIBED = 2101,
     NEU_ERR_GROUP_NOT_SUBSCRIBE      = 2102,

src/adapter/adapter.c

@@ -861,6 +861,12 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         break;
     }
+    case NEU_REQ_ADD_NDRIVER_MAP: {
+        break;
+    }
+    case NEU_REQ_DEL_NDRIVER_MAP: {
+        break;
+    }
     case NEU_REQ_UPDATE_NDRIVER_TAG_PARAM: {
         neu_req_update_ndriver_tag_param_t *cmd =
             (neu_req_update_ndriver_tag_param_t *) &header[1];

src/core/manager.c

@@ -926,9 +926,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_req_ndriver_map_t *cmd   = (neu_req_ndriver_map_t *) &header[1];
         neu_resp_error_t       error = { 0 };
 
-        // TODO
-        nlog_info("add map ndriver:`%s` driver:`%s` group:`%s`", cmd->ndriver,
-                  cmd->driver, cmd->group);
+        error.error = neu_manager_add_ndriver_map(manager, cmd->ndriver,
+                                                  cmd->driver, cmd->group);
+        if (error.error == NEU_ERR_SUCCESS) {
+            forward_msg(manager, msg, cmd->ndriver);
+            manager_storage_add_ndriver_map(manager, cmd->ndriver, cmd->driver,
+                                            cmd->group);
+        }
 
         header->type = NEU_RESP_ERROR;
         strcpy(header->receiver, header->sender);
@@ -939,9 +943,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_req_ndriver_map_t *cmd   = (neu_req_ndriver_map_t *) &header[1];
         neu_resp_error_t       error = { 0 };
 
-        // TODO
-        nlog_info("del map ndriver:`%s` driver:`%s` group:`%s`", cmd->ndriver,
-                  cmd->driver, cmd->group);
+        error.error = neu_manager_del_ndriver_map(manager, cmd->ndriver,
+                                                  cmd->driver, cmd->group);
+        if (error.error == NEU_ERR_SUCCESS) {
+            forward_msg(manager, msg, cmd->ndriver);
+            manager_storage_del_ndriver_map(manager, cmd->ndriver, cmd->driver,
+                                            cmd->group);
+        }
 
         header->type = NEU_RESP_ERROR;
         strcpy(header->receiver, header->sender);
@@ -954,10 +962,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_resp_error_t            e    = { 0 };
         neu_resp_get_ndriver_maps_t resp = { 0 };
 
-        // TODO
-        (void) cmd;
-        UT_icd icd = { sizeof(neu_resp_get_ndriver_maps_t), NULL, NULL, NULL };
-        utarray_new(resp.groups, &icd);
+        e.error =
+            neu_manager_get_ndriver_maps(manager, cmd->ndriver, &resp.groups);
 
         strcpy(header->receiver, header->sender);
         if (0 == e.error) {

src/core/manager_internal.c

@@ -572,9 +572,9 @@ UT_array *neu_manager_get_driver_group(neu_manager_t *manager)
     return driver_groups;
 }
 
-int neu_manager_subscribe(neu_manager_t *manager, const char *app,
-                          const char *driver, const char *group,
-                          const char *params)
+static inline int manager_subscribe(neu_manager_t *manager, const char *app,
+                                    const char *driver, const char *group,
+                                    const char *params)
 {
     int            ret  = NEU_ERR_SUCCESS;
     nng_pipe       pipe = { 0 };
@@ -585,25 +585,37 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
         return NEU_ERR_NODE_NOT_EXIST;
     }
 
-    if (0 == strcmp(app, "monitor")) {
-        // filter out monitor node
-        return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE;
-    }
-
     ret =
         neu_adapter_driver_group_exist((neu_adapter_driver_t *) adapter, group);
     if (ret != NEU_ERR_SUCCESS) {
         return ret;
     }
 
-    adapter = neu_node_manager_find(manager->node_manager, app);
+    pipe = neu_node_manager_get_pipe(manager->node_manager, app);
+    return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app,
+                                     group, params, pipe);
+}
+
+int neu_manager_subscribe(neu_manager_t *manager, const char *app,
+                          const char *driver, const char *group,
+                          const char *params)
+{
+    if (0 == strcmp(app, "monitor")) {
+        // filter out monitor node
+        return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE;
+    }
+
+    neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, app);
+
     if (adapter == NULL) {
         return NEU_ERR_NODE_NOT_EXIST;
     }
 
-    pipe = neu_node_manager_get_pipe(manager->node_manager, app);
-    return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app,
-                                     group, params, pipe);
+    if (NEU_NA_TYPE_APP != neu_adapter_get_type(adapter)) {
+        return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE;
+    }
+
+    return manager_subscribe(manager, app, driver, group, params);
 }
 
 int neu_manager_send_subscribe(neu_manager_t *manager, const char *app,
@@ -672,6 +684,46 @@ UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
     return subs;
 }
 
+int neu_manager_add_ndriver_map(neu_manager_t *manager, const char *ndriver,
+                                const char *driver, const char *group)
+{
+    neu_adapter_t *adapter =
+        neu_node_manager_find(manager->node_manager, ndriver);
+
+    if (adapter == NULL) {
+        return NEU_ERR_NODE_NOT_EXIST;
+    }
+
+    if (NEU_NA_TYPE_NDRIVER != neu_adapter_get_type(adapter)) {
+        return NEU_ERR_NODE_NOT_ALLOW_MAP;
+    }
+
+    return manager_subscribe(manager, ndriver, driver, group, NULL);
+}
+
+int neu_manager_del_ndriver_map(neu_manager_t *manager, const char *ndriver,
+                                const char *driver, const char *group)
+{
+    return neu_manager_unsubscribe(manager, ndriver, driver, group);
+}
+
+int neu_manager_get_ndriver_maps(neu_manager_t *manager, const char *ndriver,
+                                 UT_array **result)
+{
+    neu_adapter_t *adapter =
+        neu_node_manager_find(manager->node_manager, ndriver);
+
+    if (adapter == NULL) {
+        return NEU_ERR_NODE_NOT_EXIST;
+    } else if (NEU_NA_TYPE_NDRIVER != neu_adapter_get_type(adapter)) {
+        return NEU_ERR_NODE_NOT_ALLOW_MAP;
+    }
+
+    *result = neu_subscribe_manager_get_ndriver_maps(manager->subscribe_manager,
+                                                     ndriver);
+    return 0;
+}
+
 int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
                               neu_persist_node_info_t *info)
 {

src/core/manager_internal.h

@@ -105,6 +105,12 @@ int       neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
 UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app);
 UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
                                               const char *   app);
+int neu_manager_add_ndriver_map(neu_manager_t *manager, const char *ndriver,
+                                const char *driver, const char *group);
+int neu_manager_del_ndriver_map(neu_manager_t *manager, const char *ndriver,
+                                const char *driver, const char *group);
+int neu_manager_get_ndriver_maps(neu_manager_t *manager, const char *ndriver,
+                                 UT_array **result);
 
 int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
                               neu_persist_node_info_t *info);

src/core/storage.c

@@ -92,6 +92,20 @@ void manager_storage_unsubscribe(neu_manager_t *manager, const char *app,
     }
 }
 
+void manager_storage_add_ndriver_map(neu_manager_t *manager,
+                                     const char *ndriver, const char *driver,
+                                     const char *group)
+{
+    return manager_storage_subscribe(manager, ndriver, driver, group, NULL);
+}
+
+void manager_storage_del_ndriver_map(neu_manager_t *manager,
+                                     const char *ndriver, const char *driver,
+                                     const char *group)
+{
+    return manager_storage_unsubscribe(manager, ndriver, driver, group);
+}
+
 static int save_node_group_and_tags(neu_group_t *grp, void *data)
 {
     const char *node     = data;

src/core/storage.h

@@ -34,6 +34,12 @@ void manager_storage_subscribe(neu_manager_t *manager, const char *app,
                                const char *params);
 void manager_storage_unsubscribe(neu_manager_t *manager, const char *app,
                                  const char *driver, const char *group);
+void manager_storage_add_ndriver_map(neu_manager_t *manager,
+                                     const char *ndriver, const char *driver,
+                                     const char *group);
+void manager_storage_del_ndriver_map(neu_manager_t *manager,
+                                     const char *ndriver, const char *driver,
+                                     const char *group);
 
 void manager_storage_inst_node(neu_manager_t *manager, const char *tmpl_name,
                                const char *node);

src/core/subscribe.c

@@ -132,6 +132,30 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
     return groups;
 }
 
+UT_array *neu_subscribe_manager_get_ndriver_maps(neu_subscribe_mgr_t *mgr,
+                                                 const char *         ndriver)
+{
+    UT_array *groups = NULL;
+    UT_icd    icd = { sizeof(neu_resp_ndriver_map_info_t), NULL, NULL, NULL };
+    utarray_new(groups, &icd);
+
+    sub_elem_t *el = NULL, *tmp = NULL;
+    HASH_ITER(hh, mgr->ss, el, tmp)
+    {
+        utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
+        {
+            if (strcmp(sub_app->app_name, ndriver) == 0) {
+                neu_resp_ndriver_map_info_t info = { 0 };
+                strncpy(info.driver, el->key.driver, sizeof(info.driver));
+                strncpy(info.group, el->key.group, sizeof(info.group));
+                utarray_push_back(groups, &info);
+            }
+        }
+    }
+
+    return groups;
+}
+
 void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app)
 {
     sub_elem_t *el = NULL, *tmp = NULL;

src/core/subscribe.h

@@ -47,6 +47,8 @@ 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);
 UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app);
+UT_array *neu_subscribe_manager_get_ndriver_maps(neu_subscribe_mgr_t *mgr,
+                                                 const char *         ndriver);
 void neu_subscribe_manager_unsub_all(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,

src/parser/neu_json_ndriver.c

@@ -198,15 +198,9 @@ int neu_json_encode_ndriver_tag(void *json_obj, void *param)
         return -1;
     }
 
-    neu_json_elem_t params_elem[] = {
-        {
-            .name      = "params",
-            .t         = NEU_JSON_STR,
-            .v.val_str = tag->params,
-        },
-    };
-    ret = neu_json_encode_field(json_obj, params_elem,
-                                NEU_JSON_ELEM_SIZE(params_elem));
+    if (tag->params) {
+        ret = neu_json_load_key(json_obj, "params", tag->params, true);
+    }
 
     return ret;
 }

src/utils/http.c

@@ -324,6 +324,7 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_NODE_SETTING_INVALID:
     case NEU_ERR_NODE_NOT_ALLOW_UPDATE:
     case NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE:
+    case NEU_ERR_NODE_NOT_ALLOW_MAP:
     case NEU_ERR_LIBRARY_INFO_INVALID:
     case NEU_ERR_LICENSE_INVALID:
     case NEU_ERR_GROUP_PARAMETER_INVALID: