Merge pull request #1805 from eeff/main

feat(restful): add node api support setting

include/neuron/msg.h

@@ -238,10 +238,16 @@ typedef struct neu_resp_get_plugin {
 } neu_resp_get_plugin_t;
 
 typedef struct neu_req_add_node {
-    char node[NEU_NODE_NAME_LEN];
-    char plugin[NEU_PLUGIN_NAME_LEN];
+    char  node[NEU_NODE_NAME_LEN];
+    char  plugin[NEU_PLUGIN_NAME_LEN];
+    char *setting;
 } neu_req_add_node_t;
 
+static inline void neu_req_add_node_fini(neu_req_add_node_t *req)
+{
+    free(req->setting);
+}
+
 typedef struct neu_req_update_node {
     char node[NEU_NODE_NAME_LEN];
     char new_name[NEU_NODE_NAME_LEN];

plugins/restful/adapter_handle.c

@@ -52,8 +52,11 @@ void handle_add_adapter(nng_aio *aio)
                 header.type = NEU_REQ_ADD_NODE;
                 strcpy(cmd.node, req->name);
                 strcpy(cmd.plugin, req->plugin);
-                ret = neu_plugin_op(plugin, header, &cmd);
+                cmd.setting  = req->setting;
+                req->setting = NULL; // moved
+                ret          = neu_plugin_op(plugin, header, &cmd);
                 if (ret != 0) {
+                    neu_req_add_node_fini(&cmd);
                     NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
                         neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
                     });

src/core/manager.c

@@ -34,6 +34,7 @@
 #include "adapter.h"
 #include "adapter/adapter_internal.h"
 #include "adapter/driver/driver_internal.h"
+#include "adapter/storage.h"
 #include "argparse.h"
 #include "base/msg_internal.h"
 #include "errcodes.h"
@@ -657,14 +658,18 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     }
     case NEU_REQ_ADD_NODE: {
         neu_req_add_node_t *cmd = (neu_req_add_node_t *) &header[1];
-        int                 error =
-            neu_manager_add_node(manager, cmd->node, cmd->plugin, false, false);
+        int error = neu_manager_add_node(manager, cmd->node, cmd->plugin,
+                                         cmd->setting, false, false);
         neu_resp_error_t e = { .error = error };
 
         if (error == NEU_ERR_SUCCESS) {
             manager_storage_add_node(manager, cmd->node);
+            if (cmd->setting) {
+                adapter_storage_setting(cmd->node, cmd->setting);
+            }
         }
 
+        neu_req_add_node_fini(cmd);
         header->type = NEU_RESP_ERROR;
         strcpy(header->receiver, header->sender);
         reply(manager, header, &e);

src/core/manager_internal.c

@@ -46,7 +46,7 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager)
 }
 
 int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
-                         const char *             plugin_name,
+                         const char *plugin_name, const char *setting,
                          neu_node_running_state_e state, bool load)
 {
     neu_adapter_t *       adapter      = NULL;
@@ -86,6 +86,13 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
     neu_node_manager_add(manager->node_manager, adapter);
     neu_adapter_init(adapter, state);
 
+    if (NULL != setting &&
+        0 != (ret = neu_adapter_set_setting(adapter, setting))) {
+        neu_node_manager_del(manager->node_manager, node_name);
+        neu_adapter_uninit(adapter);
+        return ret;
+    }
+
     return NEU_ERR_SUCCESS;
 }
 

src/core/manager_internal.h

@@ -49,7 +49,7 @@ int       neu_manager_del_plugin(neu_manager_t *manager, const char *plugin);
 UT_array *neu_manager_get_plugins(neu_manager_t *manager);
 
 int       neu_manager_add_node(neu_manager_t *manager, const char *node_name,
-                               const char *             plugin_name,
+                               const char *plugin_name, const char *setting,
                                neu_node_running_state_e state, bool load);
 int       neu_manager_del_node(neu_manager_t *manager, const char *node_name);
 UT_array *neu_manager_get_nodes(neu_manager_t *manager, int type,

src/core/storage.c

@@ -140,8 +140,8 @@ int manager_load_node(neu_manager_t *manager)
     utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
     {
         rv                    = neu_manager_add_node(manager, node_info->name,
-                                  node_info->plugin_name, node_info->state,
-                                  true);
+                                  node_info->plugin_name, NULL,
+                                  node_info->state, true);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
         nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",
                     ok_or_err, node_info->type, node_info->name,

src/parser/neu_json_node.c

@@ -81,10 +81,17 @@ int neu_json_decode_add_node_req(char *buf, neu_json_add_node_req_t **result)
     req->plugin = req_elems[0].v.val_str;
     req->name   = req_elems[1].v.val_str;
 
+    ret = neu_json_dump_key(json_obj, "params", &req->setting, false);
+    if (0 != ret) {
+        goto decode_fail;
+    }
+
     *result = req;
     goto decode_exit;
 
 decode_fail:
+    free(req_elems[0].v.val_str);
+    free(req_elems[1].v.val_str);
     free(req);
     ret = -1;
 
@@ -100,6 +107,7 @@ void neu_json_decode_add_node_req_free(neu_json_add_node_req_t *req)
 
     free(req->plugin);
     free(req->name);
+    free(req->setting);
 
     free(req);
 }

src/parser/neu_json_node.h

@@ -35,6 +35,7 @@ extern "C" {
 typedef struct {
     char *name;
     char *plugin;
+    char *setting;
 } neu_json_add_node_req_t;
 
 int  neu_json_encode_add_node_req(void *json_object, void *param);

tests/ft/app/test_app.py

@@ -24,6 +24,41 @@ class TestDriver:
         assert 200 == response.status_code
         assert error.NEU_ERR_SUCCESS == response.json()['error']
 
+    @description(given="running neuron", when="add app node with wrong setting", then="should fail")
+    @pytest.mark.parametrize('node,plugin', [('mqtt-test', config.PLUGIN_MQTT)])
+    def test_create_app_node_with_wrong_setting(self, node, plugin):
+        params = {"client-id": "neuron_aBcDeF"}
+        response = api.add_node(node=node, plugin=plugin, params=params)
+        assert 400 == response.status_code
+        assert error.NEU_ERR_NODE_SETTING_INVALID == response.json()['error']
+
+        response = api.get_nodes_state(node)
+        assert 404 == response.status_code
+        assert error.NEU_ERR_NODE_NOT_EXIST == response.json()['error']
+
+    @description(given="running neuron", when="add app node with correct setting", then="should success")
+    @pytest.mark.parametrize('node,plugin', [('mqtt-test', config.PLUGIN_MQTT)])
+    def test_create_app_node_with_correct_setting(self, node, plugin):
+        params = {
+            "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,
+            "host": "broker.emqx.io",
+            "port": 1883,
+            "ssl": False
+        }
+        response = api.add_node(node=node, plugin=plugin, params=params)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.get_nodes_state(node)
+        assert 200 == response.status_code
+        assert config.NEU_NODE_STATE_RUNNING == response.json()['running']
+        api.del_node(node)
+
     @description(given="app node with subscriptions", when="delete node", then="should success")
     @pytest.mark.parametrize('app,plugin', [('mqtt', config.PLUGIN_MQTT)])
     def test_delete_app(self, app, plugin):

tests/ft/driver/test_driver.py

@@ -15,6 +15,39 @@ class TestDriver:
         assert 200 == response.status_code
         assert error.NEU_ERR_SUCCESS == response.json()['error']
 
+    @description(given="running neuron", when="add driver node with wrong setting", then="should fail")
+    @pytest.mark.parametrize('node,plugin', [('modbus-01', config.PLUGIN_MODBUS_TCP)])
+    def test_create_node_with_wrong_setting(self, node, plugin):
+        params = {"connection_mode": 0}
+        response = api.add_node(node=node, plugin=plugin, params=params)
+        assert 400 == response.status_code
+        assert error.NEU_ERR_NODE_SETTING_INVALID == response.json()['error']
+
+        response = api.get_nodes_state(node)
+        assert 404 == response.status_code
+        assert error.NEU_ERR_NODE_NOT_EXIST == response.json()['error']
+
+    @description(given="running neuron", when="add driver node with correct setting", then="should success")
+    @pytest.mark.parametrize('node,plugin', [('modbus-01', config.PLUGIN_MODBUS_TCP)])
+    def test_create_node_with_correct_setting(self, node, plugin):
+        params = {
+            "connection_mode": 0,
+            "transport_mode": 0,
+            "interval": 20,
+            "host": "127.0.0.1",
+            "port": 502,
+            "timeout": 3000,
+            "max_retries": 2
+        }
+        response = api.add_node(node=node, plugin=plugin, params=params)
+        assert 200 == response.status_code
+        assert error.NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.get_nodes_state(node)
+        assert 200 == response.status_code
+        assert config.NEU_NODE_STATE_RUNNING == response.json()['running']
+        api.del_node(node)
+
     @description(given="created node", when="get node state", then="state is init")
     @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
     def test_node_init(self, node):

tests/ft/neuron/api.py

@@ -58,8 +58,11 @@ def change_log_level(json, jwt=config.default_jwt):
 
 
 @gen_check
-def add_node(node, plugin, jwt=config.default_jwt):
-    return requests.post(url=config.BASE_URL + '/api/v2/node', headers={"Authorization": jwt}, json={"name": node, "plugin": plugin})
+def add_node(node, plugin, params=None, jwt=config.default_jwt):
+    body = {"name": node, "plugin": plugin}
+    if params:
+        body["params"] = params
+    return requests.post(url=config.BASE_URL + '/api/v2/node', headers={"Authorization": jwt}, json=body)
 
 
 @gen_check