diff --git a/include/neuron/msg.h b/include/neuron/msg.h index 086d2a2a..e1eef573 100644 --- a/include/neuron/msg.h +++ b/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]; diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index a45651b9..b281f44f 100644 --- a/plugins/restful/adapter_handle.c +++ b/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); }); diff --git a/src/core/manager.c b/src/core/manager.c index ab23316b..2ed89cbf 100644 --- a/src/core/manager.c +++ b/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); diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index f74e2b8f..f68c0d34 100644 --- a/src/core/manager_internal.c +++ b/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; } diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index e89e0b63..5df1d88c 100644 --- a/src/core/manager_internal.h +++ b/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, diff --git a/src/core/storage.c b/src/core/storage.c index 296d187c..fece57e0 100644 --- a/src/core/storage.c +++ b/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, diff --git a/src/parser/neu_json_node.c b/src/parser/neu_json_node.c index 943eebcd..360562f9 100644 --- a/src/parser/neu_json_node.c +++ b/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); } diff --git a/src/parser/neu_json_node.h b/src/parser/neu_json_node.h index 478527d0..47fdb55f 100644 --- a/src/parser/neu_json_node.h +++ b/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); diff --git a/tests/ft/app/test_app.py b/tests/ft/app/test_app.py index 59c8763a..b51d0638 100644 --- a/tests/ft/app/test_app.py +++ b/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): diff --git a/tests/ft/driver/test_driver.py b/tests/ft/driver/test_driver.py index b28bed60..da06d5e6 100644 --- a/tests/ft/driver/test_driver.py +++ b/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): diff --git a/tests/ft/neuron/api.py b/tests/ft/neuron/api.py index ee9cf9a1..14a51712 100644 --- a/tests/ft/neuron/api.py +++ b/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