Merge pull request #1020 from eeff/eeff/dev

feat: add JSON string params in group sub/unsub message

include/neuron/adapter.h

@@ -294,22 +294,35 @@ typedef struct neu_resp_get_tag {
     UT_array *tags; // array neu_datatag_t
 } neu_resp_get_tag_t;
 
+typedef struct {
+    char  app[NEU_NODE_NAME_LEN];
+    char  driver[NEU_NODE_NAME_LEN];
+    char  group[NEU_GROUP_NAME_LEN];
+    char *params;
+} neu_req_subscribe_t;
+
 typedef struct {
     char app[NEU_NODE_NAME_LEN];
     char driver[NEU_NODE_NAME_LEN];
     char group[NEU_GROUP_NAME_LEN];
-} neu_req_subscribe_t, neu_req_unsubscribe_t;
+} neu_req_unsubscribe_t;
 
 typedef struct neu_req_get_subscribe_group {
     char app[NEU_NODE_NAME_LEN];
 } neu_req_get_subscribe_group_t, neu_req_get_sub_driver_tags_t;
 
 typedef struct neu_resp_subscribe_info {
-    char app[NEU_NODE_NAME_LEN];
-    char driver[NEU_NODE_NAME_LEN];
-    char group[NEU_GROUP_NAME_LEN];
+    char  app[NEU_NODE_NAME_LEN];
+    char  driver[NEU_NODE_NAME_LEN];
+    char  group[NEU_GROUP_NAME_LEN];
+    char *params;
 } neu_resp_subscribe_info_t;
 
+static inline void neu_resp_subscribe_info_fini(neu_resp_subscribe_info_t *info)
+{
+    free(info->params);
+}
+
 typedef struct {
     char      driver[NEU_NODE_NAME_LEN];
     char      group[NEU_GROUP_NAME_LEN];

include/neuron/json/json.h

@@ -88,6 +88,9 @@ void *neu_json_encode_array_value(void *array, neu_json_elem_t *t, int n);
 int   neu_json_encode_field(void *json_object, neu_json_elem_t *elem, int n);
 int   neu_json_encode(void *json_object, char **str);
 
+int neu_json_dump_key(void *object, const char *key, char **const result,
+                      bool must_exist);
+
 #ifdef __cplusplus
 }
 #endif

plugins/restful/group_config_handle.c

@@ -208,7 +208,9 @@ void handle_grp_subscribe(nng_aio *aio)
             strcpy(cmd.app, req->app);
             strcpy(cmd.driver, req->driver);
             strcpy(cmd.group, req->group);
-            ret = neu_plugin_op(plugin, header, &cmd);
+            cmd.params  = req->params; // ownership moved
+            req->params = NULL;
+            ret         = neu_plugin_op(plugin, header, &cmd);
             if (ret != 0) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
                     neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
@@ -284,6 +286,7 @@ void handle_grp_get_subscribe_resp(nng_aio *                       aio,
         int index = utarray_eltidx(groups->groups, group);
         sub_grp_configs.groups[index].driver = group->driver;
         sub_grp_configs.groups[index].group  = group->group;
+        sub_grp_configs.groups[index].params = group->params;
     }
 
     neu_json_encode_by_fn(&sub_grp_configs, neu_json_encode_get_subscribe_resp,

src/adapter/adapter.c

@@ -385,6 +385,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_RESP_GET_NODE_STATE:
     case NEU_RESP_GET_NODES_STATE:
     case NEU_RESP_GET_NODE_SETTING:
+    case NEU_REQ_SUBSCRIBE_GROUP:
+    case NEU_REQ_UNSUBSCRIBE_GROUP:
     case NEU_RESP_READ_GROUP:
     case NEU_RESP_GET_SUBSCRIBE_GROUP:
     case NEU_RESP_ADD_TAG:

src/core/manager.c

@@ -400,10 +400,11 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_req_subscribe_t *cmd   = (neu_req_subscribe_t *) &header[1];
         neu_resp_error_t     error = { 0 };
 
-        error.error =
-            neu_manager_subscribe(manager, cmd->app, cmd->driver, cmd->group);
+        error.error = neu_manager_subscribe(manager, cmd->app, cmd->driver,
+                                            cmd->group, cmd->params);
 
         if (error.error == NEU_ERR_SUCCESS) {
+            forward_msg(manager, msg, cmd->app);
             manager_storage_subscribe(manager, cmd->app, cmd->driver,
                                       cmd->group);
         }
@@ -421,6 +422,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             neu_manager_unsubscribe(manager, cmd->app, cmd->driver, cmd->group);
 
         if (error.error == NEU_ERR_SUCCESS) {
+            forward_msg(manager, msg, cmd->app);
             manager_storage_unsubscribe(manager, cmd->app, cmd->driver,
                                         cmd->group);
         }
@@ -433,7 +435,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_GET_SUBSCRIBE_GROUP: {
         neu_req_get_subscribe_group_t *cmd =
             (neu_req_get_subscribe_group_t *) &header[1];
-        UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
+        UT_array *groups =
+            neu_manager_get_sub_group_deep_copy(manager, cmd->app);
         neu_resp_get_subscribe_group_t resp = { .groups = groups };
 
         strcpy(header->receiver, header->sender);

src/core/manager_internal.c

@@ -139,7 +139,8 @@ UT_array *neu_manager_get_driver_group(neu_manager_t *manager)
 }
 
 int neu_manager_subscribe(neu_manager_t *manager, const char *app,
-                          const char *driver, const char *group)
+                          const char *driver, const char *group,
+                          const char *params)
 {
     int            ret  = NEU_ERR_SUCCESS;
     nng_pipe       pipe = { 0 };
@@ -168,7 +169,7 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
 
     pipe = neu_node_manager_get_pipe(manager->node_manager, app);
     return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app,
-                                     group, pipe);
+                                     group, params, pipe);
 }
 
 int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
@@ -183,6 +184,24 @@ UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app)
     return neu_subscribe_manager_get(manager->subscribe_manager, app);
 }
 
+UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
+                                              const char *   app)
+{
+    UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app);
+
+    utarray_foreach(subs, neu_resp_subscribe_info_t *, sub)
+    {
+        if (sub->params) {
+            sub->params = strdup(sub->params);
+        }
+    }
+
+    // set vector element destructor
+    subs->icd.dtor = (void (*)(void *)) neu_resp_subscribe_info_fini;
+
+    return subs;
+}
+
 int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
                               neu_persist_node_info_t *info)
 {

src/core/manager_internal.h

@@ -58,10 +58,13 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type,
 UT_array *neu_manager_get_driver_group(neu_manager_t *manager);
 
 int       neu_manager_subscribe(neu_manager_t *manager, const char *app,
-                                const char *driver, const char *group);
+                                const char *driver, const char *group,
+                                const char *params);
 int       neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
                                   const char *driver, const char *group);
 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_get_node_info(neu_manager_t *manager, const char *name,
                               neu_persist_node_info_t *info);

src/core/storage.c

@@ -147,7 +147,8 @@ int manager_load_subscribe(neu_manager_t *manager)
             utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info)
             {
                 rv = neu_manager_subscribe(manager, node->node,
-                                           info->driver_name, info->group_name);
+                                           info->driver_name, info->group_name,
+                                           NULL);
                 const char *ok_or_err = (0 == rv) ? "success" : "fail";
                 nlog_info("%s load subscription app:%s driver:%s grp:%s",
                           ok_or_err, node->node, info->driver_name,

src/core/subscribe.c

@@ -58,6 +58,10 @@ void neu_subscribe_manager_destroy(neu_subscribe_mgr_t *mgr)
     HASH_ITER(hh, mgr->ss, el, tmp)
     {
         HASH_DEL(mgr->ss, el);
+        utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
+        {
+            neu_app_subscribe_fini(sub_app);
+        }
         utarray_free(el->apps);
         free(el);
     }
@@ -118,6 +122,7 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
                 strncpy(info.driver, el->key.driver, sizeof(info.driver));
                 strncpy(info.app, app, sizeof(info.app));
                 strncpy(info.group, el->key.group, sizeof(info.group));
+                info.params = sub_app->params; // borrowed reference
 
                 utarray_push_back(groups, &info);
             }
@@ -145,7 +150,8 @@ 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, nng_pipe pipe)
+                              const char *app, const char *group,
+                              const char *params, nng_pipe pipe)
 {
     sub_elem_t *        find    = NULL;
     sub_elem_key_t      key     = { 0 };
@@ -156,12 +162,17 @@ int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver,
     strncpy(app_sub.app_name, app, sizeof(app_sub.app_name));
     app_sub.pipe = pipe;
 
+    if (params && NULL == (app_sub.params = strdup(params))) {
+        return NEU_ERR_EINTERNAL;
+    }
+
     HASH_FIND(hh, mgr->ss, &key, sizeof(sub_elem_key_t), find);
 
     if (find) {
         utarray_foreach(find->apps, neu_app_subscribe_t *, sub)
         {
             if (strcmp(sub->app_name, app) == 0) {
+                neu_app_subscribe_fini(&app_sub);
                 return NEU_ERR_GROUP_ALREADY_SUBSCRIBED;
             }
         }
@@ -191,6 +202,7 @@ int neu_subscribe_manager_unsub(neu_subscribe_mgr_t *mgr, const char *driver,
         utarray_foreach(find->apps, neu_app_subscribe_t *, sub)
         {
             if (strcmp(sub->app_name, app) == 0) {
+                neu_app_subscribe_fini(sub);
                 utarray_erase(find->apps, utarray_eltidx(find->apps, sub), 1);
                 return NEU_ERR_SUCCESS;
             }
@@ -210,6 +222,10 @@ void neu_subscribe_manager_remove(neu_subscribe_mgr_t *mgr, const char *driver,
         if (strcmp(driver, el->key.driver) == 0 &&
             (group == NULL || strcmp(group, el->key.group) == 0)) {
             HASH_DEL(mgr->ss, el);
+            utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
+            {
+                neu_app_subscribe_fini(sub_app);
+            }
             utarray_free(el->apps);
             free(el);
         }

src/core/subscribe.h

@@ -29,9 +29,15 @@ typedef struct neu_subscribe_mgr neu_subscribe_mgr_t;
 
 typedef struct neu_app_subscribe {
     char     app_name[NEU_NODE_NAME_LEN];
+    char *   params;
     nng_pipe pipe;
 } neu_app_subscribe_t;
 
+static inline void neu_app_subscribe_fini(neu_app_subscribe_t *app_sub)
+{
+    free(app_sub->params);
+}
+
 neu_subscribe_mgr_t *neu_subscribe_manager_create();
 void                 neu_subscribe_manager_destroy(neu_subscribe_mgr_t *mgr);
 
@@ -44,7 +50,7 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app);
 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,
-                               nng_pipe pipe);
+                               const char *params, nng_pipe pipe);
 int  neu_subscribe_manager_unsub(neu_subscribe_mgr_t *mgr, const char *driver,
                                  const char *app, const char *group);
 void neu_subscribe_manager_remove(neu_subscribe_mgr_t *mgr, const char *driver,

src/parser/neu_json_group_config.c

@@ -25,6 +25,8 @@
 #include <stdlib.h>
 #include <string.h>
 
+#include <jansson.h>
+
 #include "json/json.h"
 
 #include "neu_json_group_config.h"
@@ -231,7 +233,7 @@ int neu_json_encode_get_driver_group_resp(void *json_object, void *param)
     return ret;
 }
 
-int neu_json_encode_get_subscribe_resp(void *json_object, void *param)
+int neu_json_encode_get_subscribe_resp(void *object, void *param)
 {
     int                            ret = 0;
     neu_json_get_subscribe_resp_t *resp =
@@ -240,18 +242,21 @@ int neu_json_encode_get_subscribe_resp(void *json_object, void *param)
     void *                               group_array = neu_json_array();
     neu_json_get_subscribe_resp_group_t *p_group     = resp->groups;
     for (int i = 0; i < resp->n_group; i++) {
-        neu_json_elem_t group_elems[] = { {
-                                              .name      = "driver",
-                                              .t         = NEU_JSON_STR,
-                                              .v.val_str = p_group->driver,
-                                          },
-                                          {
-                                              .name      = "group",
-                                              .t         = NEU_JSON_STR,
-                                              .v.val_str = p_group->group,
-                                          } };
-        group_array = neu_json_encode_array(group_array, group_elems,
-                                            NEU_JSON_ELEM_SIZE(group_elems));
+        json_t *ob = json_object();
+
+        // rely on lib jansson to check for NULL pointers
+        json_object_set_new(ob, "driver", json_string(p_group->driver));
+        json_object_set_new(ob, "group", json_string(p_group->group));
+
+        if (p_group->params) {
+            json_t *t = json_loads(p_group->params, 0, NULL);
+            json_t *p = json_object_get(t, "params");
+            json_object_set(ob, "params", p);
+            json_decref(t);
+        }
+
+        json_array_append_new(group_array, ob);
+
         p_group++;
     }
 
@@ -260,12 +265,17 @@ int neu_json_encode_get_subscribe_resp(void *json_object, void *param)
         .t            = NEU_JSON_OBJECT,
         .v.val_object = group_array,
     } };
-    ret = neu_json_encode_field(json_object, resp_elems,
+    ret                          = neu_json_encode_field(object, resp_elems,
                                 NEU_JSON_ELEM_SIZE(resp_elems));
 
     return ret;
 }
 
+static inline int dump_params(void *root, char **const result)
+{
+    return neu_json_dump_key(root, "params", result, false);
+}
+
 int neu_json_decode_subscribe_req(char *buf, neu_json_subscribe_req_t **result)
 {
     int                       ret      = 0;
@@ -299,6 +309,11 @@ int neu_json_decode_subscribe_req(char *buf, neu_json_subscribe_req_t **result)
     req->group  = req_elems[1].v.val_str;
     req->driver = req_elems[2].v.val_str;
 
+    ret = dump_params(json_obj, &req->params);
+    if (0 != ret) {
+        goto decode_fail;
+    }
+
     *result = req;
     goto decode_exit;
 
@@ -318,6 +333,7 @@ void neu_json_decode_subscribe_req_free(neu_json_subscribe_req_t *req)
     free(req->app);
     free(req->driver);
     free(req->group);
+    free(req->params);
 
     free(req);
 }

src/parser/neu_json_group_config.h

@@ -82,6 +82,7 @@ int neu_json_encode_get_driver_group_resp(void *json_object, void *param);
 typedef struct {
     char *group;
     char *driver;
+    char *params;
 } neu_json_get_subscribe_resp_group_t;
 
 typedef struct {
@@ -95,6 +96,7 @@ typedef struct {
     char *group;
     char *app;
     char *driver;
+    char *params;
 } neu_json_subscribe_req_t;
 
 int neu_json_decode_subscribe_req(char *buf, neu_json_subscribe_req_t **result);

src/utils/json.c

@@ -406,3 +406,29 @@ int neu_json_decode_value(void *object, neu_json_elem_t *ele)
 {
     return decode_object(object, ele);
 }
+
+int neu_json_dump_key(void *object, const char *key, char **const result,
+                      bool must_exist)
+{
+    int rv        = 0;
+    json_t *new   = NULL;
+    json_t *value = NULL;
+
+    value = json_object_get(object, key);
+    if (NULL == value) {
+        if (must_exist) {
+            return -1;
+        } else {
+            *result = NULL;
+            return 0;
+        }
+    }
+
+    if (NULL == (new = json_object()) || json_object_set(new, key, value) < 0 ||
+        NULL == (*result = json_dumps(new, JSON_COMPACT))) {
+        rv = -1;
+    }
+
+    json_decref(new);
+    return rv;
+}