Merge pull request #1797 from eeff/eeff/dev

feat(core): get subscribe group request support filter

include/neuron/msg.h

@@ -457,7 +457,13 @@ neu_req_subscribe_groups_fini(neu_req_subscribe_groups_t *req)
 
 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;
+    char driver[NEU_NODE_NAME_LEN];
+    char group[NEU_GROUP_NAME_LEN];
+} neu_req_get_subscribe_group_t;
+
+typedef struct {
+    char app[NEU_NODE_NAME_LEN];
+} neu_req_get_sub_driver_tags_t;
 
 typedef struct neu_resp_subscribe_info {
     char  app[NEU_NODE_NAME_LEN];

plugins/restful/group_config_handle.c

@@ -356,23 +356,41 @@ void handle_grp_get_subscribe(nng_aio *aio)
 {
     int                           ret    = 0;
     neu_plugin_t *                plugin = neu_rest_get_plugin();
-    char                          node_name[NEU_NODE_NAME_LEN] = { 0 };
-    neu_req_get_subscribe_group_t cmd                          = { 0 };
-    neu_reqresp_head_t            header                       = {
+    neu_req_get_subscribe_group_t cmd    = { 0 };
+    neu_reqresp_head_t            header = {
         .ctx  = aio,
         .type = NEU_REQ_GET_SUBSCRIBE_GROUP,
     };
 
     NEU_VALIDATE_JWT(aio);
 
-    if (neu_http_get_param_str(aio, "app", node_name, sizeof(node_name)) <= 0) {
+    // required parameter
+    ret = neu_http_get_param_str(aio, "app", cmd.app, sizeof(cmd.app));
+    if (ret <= 0 || (size_t) ret == sizeof(cmd.driver)) {
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
+            neu_http_response(aio, error_code.error, result_error);
+        })
+        return;
+    }
+
+    // optional parameter
+    ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
+    if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
+            neu_http_response(aio, error_code.error, result_error);
+        })
+        return;
+    }
+
+    // optional parameter
+    ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
+    if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
         NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
             neu_http_response(aio, error_code.error, result_error);
         })
         return;
     }
 
-    strcpy(cmd.app, node_name);
     ret = neu_plugin_op(plugin, header, &cmd);
     if (ret != 0) {
         NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {

src/core/manager.c

@@ -722,7 +722,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         manager_storage_del_node(manager, cmd->node);
         if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_APP) {
             UT_array *subscriptions = neu_subscribe_manager_get(
-                manager->subscribe_manager, cmd->node);
+                manager->subscribe_manager, cmd->node, NULL, NULL);
             neu_subscribe_manager_unsub_all(manager->subscribe_manager,
                                             cmd->node);
 
@@ -887,8 +887,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_deep_copy(manager, cmd->app);
+        UT_array *groups = neu_manager_get_sub_group_deep_copy(
+            manager, cmd->app, cmd->driver, cmd->group);
         neu_resp_get_subscribe_group_t resp = { .groups = groups };
 
         strcpy(header->receiver, header->sender);

src/core/manager_internal.c

@@ -328,13 +328,17 @@ 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)
 {
-    return neu_subscribe_manager_get(manager->subscribe_manager, app);
+    return neu_subscribe_manager_get(manager->subscribe_manager, app, NULL,
+                                     NULL);
 }
 
 UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
-                                              const char *   app)
+                                              const char *   app,
+                                              const char *   driver,
+                                              const char *   group)
 {
-    UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app);
+    UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app,
+                                               driver, group);
 
     utarray_foreach(subs, neu_resp_subscribe_info_t *, sub)
     {

src/core/manager_internal.h

@@ -74,7 +74,9 @@ 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);
+                                              const char *   app,
+                                              const char *   driver,
+                                              const char *   group);
 
 int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
                               neu_persist_node_info_t *info);

src/core/subscribe.c

@@ -105,7 +105,8 @@ UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
     return apps;
 }
 
-UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
+UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app,
+                                    const char *driver, const char *group)
 {
     sub_elem_t *el = NULL, *tmp = NULL;
     UT_array *  groups = NULL;
@@ -116,7 +117,9 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
     {
         utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
         {
-            if (strcmp(sub_app->app_name, app) == 0) {
+            if (strcmp(sub_app->app_name, app) == 0 &&
+                (!driver || strstr(el->key.driver, driver)) &&
+                (!group || strstr(el->key.group, group))) {
                 neu_resp_subscribe_info_t info = { 0 };
 
                 strncpy(info.driver, el->key.driver, sizeof(info.driver));

src/core/subscribe.h

@@ -47,7 +47,8 @@ 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);
+UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app,
+                                    const char *driver, const char *group);
 size_t    neu_subscribe_manager_group_count(const neu_subscribe_mgr_t *mgr,
                                             const char *               app);
 void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app);

tests/ft/http_api/test_http_api.py

@@ -188,6 +188,103 @@ class TestHttp:
         assert len(config_data['settings']) == 1
         assert config_data['settings'][0]['node'] == 'mqtt'
 
+    @description(given="running neuron", when="get subscribe info", then="should success")
+    def test_get_subscribe_info(self):
+        app = 'MQTT-1'
+        params = {'topic':'/neuron/test'}
+        driver_1 = 'modbus-01'
+        driver_2 = 'modbus-02'
+        group_1 = 'group-01'
+        group_2 = 'group-02'
+        try:
+            api.add_node_check(app, 'MQTT')
+            api.add_node_check(driver_1, 'Modbus TCP')
+            api.add_node_check(driver_2, 'Modbus TCP')
+            api.add_group_check(driver_1, group_1, 1000)
+            api.add_group_check(driver_1, group_2, 1000)
+            api.add_group_check(driver_2, group_1, 1000)
+            api.add_group_check(driver_2, group_2, 1000)
+            api.subscribe_group_check(app, driver_1, group_1, params=params)
+            api.subscribe_group_check(app, driver_1, group_2, params=params)
+            api.subscribe_group_check(app, driver_2, group_1, params=params)
+            api.subscribe_group_check(app, driver_2, group_2, params=params)
+
+            # match all drivers
+            response = api.get_subscribe_group(app, driver='modbus')
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 4
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_1
+            assert groups[1]['group'] == group_2
+            assert groups[2]['driver'] == driver_2
+            assert groups[2]['group'] == group_1
+            assert groups[3]['driver'] == driver_2
+            assert groups[3]['group'] == group_2
+
+            # match one driver
+            response = api.get_subscribe_group(app, driver=driver_1)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 2
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_1
+            assert groups[1]['group'] == group_2
+
+            # match none driver
+            response = api.get_subscribe_group(app, driver='none')
+            assert 200 == response.status_code
+            assert len(response.json()['groups']) == 0
+
+            # match all groups
+            response = api.get_subscribe_group(app, group='group')
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 4
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_1
+            assert groups[1]['group'] == group_2
+            assert groups[2]['driver'] == driver_2
+            assert groups[2]['group'] == group_1
+            assert groups[3]['driver'] == driver_2
+            assert groups[3]['group'] == group_2
+
+            # match one group
+            response = api.get_subscribe_group(app, group=group_1)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 2
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_2
+            assert groups[1]['group'] == group_1
+
+            # match none group
+            response = api.get_subscribe_group(app, group='none')
+            assert 200 == response.status_code
+            assert len(response.json()['groups']) == 0
+
+            # match driver and group
+            response = api.get_subscribe_group(app, driver=driver_1, group=group_1)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 1
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            response = api.get_subscribe_group(app, driver=driver_2, group=group_2)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 1
+            assert groups[0]['driver'] == driver_2
+            assert groups[0]['group'] == group_2
+        finally:
+            api.del_node(app)
+            api.del_node(driver_1)
+            api.del_node(driver_2)
+
     @description(given="running neuron", when="get mqtt shcema", then="success")
     def test_get_plugin_schema(self):
         response = api.get_plugin_schema(plugin='mqtt')

tests/ft/neuron/api.py

@@ -222,6 +222,10 @@ def get_plugins():
     return requests.get(url=config.BASE_URL + "/api/v2/plugin", headers={"Authorization": config.default_jwt})
 
 
+def get_subscribe_group(app, driver=None, group=None):
+    return requests.get(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, params={"app": app, "driver": driver, "group": group})
+
+
 @gen_check
 def subscribe_group(app, driver, group, params={}):
     return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})