Merge pull request #1655 from hxy7yx/main-1

add subscription group count to node(app) state interface

include/neuron/msg.h

@@ -625,6 +625,8 @@ typedef struct neu_resp_get_node_state {
     neu_node_state_t state;
     uint16_t         rtt; // round trip time in milliseconds
     int              core_level;
+    uint16_t         sub_group_count;
+    bool             is_driver;
 } neu_resp_get_node_state_t;
 
 typedef struct neu_req_get_nodes_state {
@@ -634,6 +636,8 @@ typedef struct {
     char             node[NEU_NODE_NAME_LEN];
     neu_node_state_t state;
     uint16_t         rtt; // round trip time in milliseconds
+    uint16_t         sub_group_count;
+    bool             is_driver;
 } neu_nodes_state_t;
 
 inline static UT_icd neu_nodes_state_t_icd()

plugins/restful/adapter_handle.c

@@ -330,11 +330,13 @@ void handle_get_node_state_resp(nng_aio *aio, neu_resp_get_node_state_t *state)
     neu_json_get_node_state_resp_t res    = { 0 };
     char *                         result = NULL;
 
-    res.link       = state->state.link;
-    res.running    = state->state.running;
-    res.rtt        = state->rtt;
-    res.log_level  = (char *) log_level_to_str(state->state.log_level);
-    res.core_level = (char *) log_level_to_str(state->core_level);
+    res.link            = state->state.link;
+    res.running         = state->state.running;
+    res.rtt             = state->rtt;
+    res.sub_group_count = state->sub_group_count;
+    res.is_driver       = state->is_driver;
+    res.log_level       = (char *) log_level_to_str(state->state.log_level);
+    res.core_level      = (char *) log_level_to_str(state->core_level);
 
     neu_json_encode_by_fn(&res, neu_json_encode_get_node_state_resp, &result);
 
@@ -359,6 +361,8 @@ void handle_get_nodes_state_resp(nng_aio *                   aio,
         states_res.nodes[index].running = state->state.running;
         states_res.nodes[index].link    = state->state.link;
         states_res.nodes[index].rtt     = state->rtt;
+        states_res.nodes[index].sub_group_count = state->sub_group_count;
+        states_res.nodes[index].is_driver       = state->is_driver;
         states_res.nodes[index].log_level =
             (char *) log_level_to_str(state->state.log_level);
     }

src/core/manager.c

@@ -1161,9 +1161,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_GET_NODES_STATE: {
         neu_resp_get_nodes_state_t resp = { 0 };
         UT_array *states = neu_node_manager_get_state(manager->node_manager);
-
-        resp.core_level = manager->log_level;
+        neu_nodes_state_t *p_state = NULL;
+        while (
+            (p_state = (neu_nodes_state_t *) utarray_next(states, p_state))) {
+            p_state->sub_group_count = neu_subscribe_manager_group_count(
+                manager->subscribe_manager, p_state->node);
+        }
         resp.states     = utarray_clone(states);
+        resp.core_level = manager->log_level;
 
         strcpy(header->receiver, header->sender);
         strcpy(header->sender, "manager");
@@ -1436,7 +1441,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_resp_get_node_state_t *resp =
             (neu_resp_get_node_state_t *) &header[1];
 
+        neu_adapter_t *adapter_node =
+            neu_node_manager_find(manager->node_manager, header->sender);
+        resp->sub_group_count = neu_subscribe_manager_group_count(
+            manager->subscribe_manager, header->sender);
+        resp->is_driver =
+            (adapter_node->module->type == NEU_NA_TYPE_DRIVER) ? true : false;
         resp->core_level = manager->log_level;
+
         forward_msg(manager, header, header->receiver);
         break;
     }

src/core/node_manager.c

@@ -415,6 +415,9 @@ UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr)
             state.state.link =
                 neu_plugin_to_plugin_common(el->adapter->plugin)->link_state;
             state.state.log_level = el->adapter->log_level;
+            state.is_driver = (el->adapter->module->type == NEU_NA_TYPE_DRIVER)
+                ? true
+                : false;
             neu_metric_entry_t *e = NULL;
             if (NULL != el->adapter->metrics) {
                 HASH_FIND_STR(el->adapter->metrics->entries,

src/core/subscribe.c

@@ -132,6 +132,25 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
     return groups;
 }
 
+size_t neu_subscribe_manager_group_count(const neu_subscribe_mgr_t *mgr,
+                                         const char *               app)
+{
+    size_t      group_count = 0;
+    sub_elem_t *current_element, *tmp;
+
+    HASH_ITER(hh, mgr->ss, current_element, tmp)
+    {
+        utarray_foreach(current_element->apps, neu_app_subscribe_t *, sub_app)
+        {
+            if (strcmp(sub_app->app_name, app) == 0) {
+                group_count++;
+            }
+        }
+    }
+
+    return group_count;
+}
+
 UT_array *neu_subscribe_manager_get_ndriver_maps(neu_subscribe_mgr_t *mgr,
                                                  const char *         ndriver)
 {

src/core/subscribe.h

@@ -48,6 +48,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);
+size_t    neu_subscribe_manager_group_count(const 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);

src/parser/neu_json_node.c

@@ -169,6 +169,7 @@ int neu_json_encode_get_node_state_resp(void *json_object, void *param)
     int                             ret = 0;
     neu_json_get_node_state_resp_t *resp =
         (neu_json_get_node_state_resp_t *) param;
+    int array_size = resp->is_driver ? 5 : 6;
 
     neu_json_elem_t resp_elems[] = { {
                                          .name      = "running",
@@ -194,9 +195,14 @@ int neu_json_encode_get_node_state_resp(void *json_object, void *param)
                                          .name      = "neuron_core",
                                          .t         = NEU_JSON_STR,
                                          .v.val_str = resp->core_level,
+                                     },
+                                     {
+                                         .name      = "sub_group_count",
+                                         .t         = NEU_JSON_INT,
+                                         .v.val_int = resp->sub_group_count,
                                      } };
-    ret = neu_json_encode_field(json_object, resp_elems,
-                                NEU_JSON_ELEM_SIZE(resp_elems));
+
+    ret = neu_json_encode_field(json_object, resp_elems, array_size);
 
     return ret;
 }
@@ -210,6 +216,7 @@ int neu_json_encode_get_nodes_state_resp(void *json_object, void *param)
     void *                      node_array = neu_json_array();
     neu_json_get_nodes_state_t *p_node     = resp->nodes;
     for (int i = 0; i < resp->n_node; i++) {
+        int             array_size   = p_node->is_driver ? 5 : 6;
         neu_json_elem_t node_elems[] = { {
                                              .name      = "node",
                                              .t         = NEU_JSON_STR,
@@ -234,9 +241,15 @@ int neu_json_encode_get_nodes_state_resp(void *json_object, void *param)
                                              .name      = "log_level",
                                              .t         = NEU_JSON_STR,
                                              .v.val_str = p_node->log_level,
+                                         },
+                                         {
+                                             .name = "sub_group_count",
+                                             .t    = NEU_JSON_INT,
+                                             .v.val_int =
+                                                 p_node->sub_group_count,
                                          } };
-        node_array = neu_json_encode_array(node_array, node_elems,
-                                           NEU_JSON_ELEM_SIZE(node_elems));
+
+        node_array = neu_json_encode_array(node_array, node_elems, array_size);
         p_node++;
     }
 

src/parser/neu_json_node.h

@@ -50,21 +50,25 @@ int  neu_json_decode_del_node_req(char *buf, neu_json_del_node_req_t **result);
 void neu_json_decode_del_node_req_free(neu_json_del_node_req_t *req);
 
 typedef struct {
-    int64_t running;
-    int64_t link;
-    int64_t rtt;
-    char *  log_level;
-    char *  core_level;
+    int64_t  running;
+    int64_t  link;
+    int64_t  rtt;
+    uint16_t sub_group_count;
+    bool     is_driver;
+    char *   log_level;
+    char *   core_level;
 } neu_json_get_node_state_resp_t;
 
 int neu_json_encode_get_node_state_resp(void *json_object, void *param);
 
 typedef struct {
-    char *  name;
-    int64_t running;
-    int64_t link;
-    int64_t rtt;
-    char *  log_level;
+    char *   name;
+    int64_t  running;
+    int64_t  link;
+    int64_t  rtt;
+    uint16_t sub_group_count;
+    bool     is_driver;
+    char *   log_level;
 } neu_json_get_nodes_state_t;
 
 typedef struct {

src/parser/neu_json_tag.c

@@ -470,7 +470,6 @@ int neu_json_decode_gtag_json(void *json_obj, neu_json_gtag_t *gtag_p)
     goto decode_exit;
 
 decode_fail:
-    free(gtag_p);
     free(gtag_elems[0].v.val_str);
     return -1;
 

tests/ft/app/__init__.py

@@ -0,0 +1,3 @@
+import sys
+
+sys.path.append("tests/ft")

tests/ft/app/test_app.py

@@ -0,0 +1,97 @@
+import neuron.api as api
+import neuron.error as error
+import neuron.config as config
+from neuron.common import *
+
+hold_int16 = [{"name": "hold_int16", "address": "1!400001",
+               "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_INT16}]
+
+class TestDriver:
+
+    @description(given="running neuron", when="add driver node", then="add success")
+    @pytest.mark.parametrize('node,plugin', [('modbus-tcp-1', config.PLUGIN_MODBUS_TCP)])
+    def test_create_driver_node(self, node, plugin):
+        response = api.add_node(node=node,
+                                plugin=plugin)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+    @description(given="running neuron", when="add app node", then="add success")
+    @pytest.mark.parametrize('node,plugin', [('mqtt-1', config.PLUGIN_MQTT)])
+    def test_create_app_node(self, node, plugin):
+        response = api.add_node(node=node,
+                                plugin=plugin)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+    @description(given="modbus tcp node", when="setting modbus tcp node and groups/tags", then="setting success")
+    def test_modbus_tcp_setting(self):
+        response = api.modbus_tcp_node_setting(node='modbus-tcp-1', port=502)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.add_group(node='modbus-tcp-1', group="group-1")
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.add_group(node='modbus-tcp-1', group="group-2")
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.add_group(node='modbus-tcp-1', group="group-3")
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.add_tags(node='modbus-tcp-1', group="group-1", tags=hold_int16)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.add_tags(node='modbus-tcp-1', group="group-2", tags=hold_int16)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.add_tags(node='modbus-tcp-1', group="group-3", tags=hold_int16)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+    @description(given="mqtt node", when="setting mqtt node and subscribe modbus group", then="setting&subscribe success")
+    def test_mqtt_setting(self):
+        response = api.mqtt_node_setting(node='mqtt-1')
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.subscribe_groups(app='mqtt-1', groups=[{"driver": "modbus-tcp-1", "group": "group-1"},
+                                                              {"driver": "modbus-tcp-1", "group": "group-2"},
+                                                              {"driver": "modbus-tcp-1", "group": "group-3"}])
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+    @description(given="mqtt node", when="subscribe/unsubscribe modbus group", then="return correct sub_group_count in node state")
+    def test_mqtt_sub_group_count(self):
+        response = api.get_nodes_state(node='mqtt-1')
+        assert 200 == response.status_code
+        assert 3 == response.json()['sub_group_count']
+
+        response = api.unsubscribe_group(app='mqtt-1', driver='modbus-tcp-1', group='group-1')
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.unsubscribe_group(app='mqtt-1', driver='modbus-tcp-1', group='group-2')
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.unsubscribe_group(app='mqtt-1', driver='modbus-tcp-1', group='group-3')
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.get_nodes_state(node='mqtt-1')
+        assert 200 == response.status_code
+        assert 0 == response.json()['sub_group_count']
+
+        response = api.subscribe_group(app='mqtt-1', driver='modbus-tcp-1', group='group-1')
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.get_nodes_state(node='mqtt-1')
+        assert 200 == response.status_code
+        assert 1 == response.json()['sub_group_count']

tests/ft/neuron/api.py

@@ -157,7 +157,6 @@ def read_tag(node, group, tag, sync=False):
     finally:
         return x[0]['value']
 
-
 def read_tag_error(node, group, tag, sync=False):
     response = read_tags(node, group, sync)
     assert 200 == response.status_code
@@ -282,6 +281,14 @@ def del_template_tags(tmpl, group, tags):
 def get_template_tags(tmpl, group, name):
     return requests.get(url=config.BASE_URL + "/api/v2/template/tag", headers={"Authorization": config.default_jwt}, params={"template": tmpl, "group": group, "name": name})
 
+def subscribe_group(app, driver, group):
+    return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group})
+
+def subscribe_groups(app, groups):
+    return requests.post(url=config.BASE_URL + "/api/v2/subscribes", headers={"Authorization": config.default_jwt}, json={"app": app, "groups": groups})
+
+def unsubscribe_group(app, driver, group):
+    return requests.delete(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group})
 
 def get_metrics(category="global", node=""):
     return requests.get(url=config.BASE_URL + "/api/v2/metrics?category=" + category + "&node=" + node, headers={"Authorization": config.default_jwt})
@@ -299,3 +306,8 @@ def modbus_rtu_node_setting(node, port, connection_mode=0, transport_mode=0, int
                                     "interval": interval, "host": host, "port": port, "timeout": timeout,
                                     "max_retries": max_retries, "link": link, "device": device, "stop": stop,
                                     "parity": parity, "baud": baud, "data": data})
+
+def mqtt_node_setting(node):
+    return node_setting(node, json={"client-id": "neuron_aBcDeF", "qos":0, "format": 0,
+                                    "write-req-topic": f"/neuron/{node}/write/req", "write-resp-topic": f"/neuron/{node}/write/resp",
+                                    "offline-cache": False, "cache-sync-interval": 100, "host": "broker.emqx.io", "port": 1883, "username": "", "password": "", "ssl": False})

tests/ft/neuron/config.py

@@ -68,3 +68,4 @@ NEU_DATA_8 = 3
 
 PLUGIN_MODBUS_TCP = "Modbus TCP"
 PLUGIN_MODBUS_RTU = "Modbus RTU"
+PLUGIN_MQTT = "MQTT"