GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-28 22:59:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #727 from fengzeroz/main

Browse Source 
fix adapter start & not allow create node with single plugin
This commit is contained in:
fengzeroz
2022-08-23 14:43:15 +08:00
committed by GitHub
parent 00a55af7da a0ab69dba1
commit a6d593eb8b
14 changed files with 55 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/resource/api.resource
-1
View File
@@ -10,7 +10,6 @@ ${PLUGIN-EKUIPER} ekuiper
${NODE_DRIVER} 1
${NODE_APP} 2
${NODE_STATE_IDLE} 0
${NODE_STATE_INIT} 1
${NODE_STATE_READY} 2
${NODE_STATE_RUNNING} 3
ft/resource/error.resource
+1
View File
@@ -47,6 +47,7 @@ ${NEU_ERR_LIBRARY_NAME_CONFLICT} 2303
${NEU_ERR_LIBRARY_FAILED_TO_OPEN} 2304
${NEU_ERR_LIBRARY_MODULE_INVALID} 2305
${NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL} 2306
${NEU_ERR_LIBRARY_NOT_ALLOW_CREATE_INSTANCE} 2307
${NEU_ERR_LICENSE_NOT_FOUND} 2400
${NEU_ERR_LICENSE_INVALID} 2401
include/neuron/adapter.h
+2
View File
@@ -175,6 +175,7 @@ typedef struct neu_resp_error {
typedef struct {
char node[NEU_NODE_NAME_LEN];
bool auto_start;
} neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t;
typedef struct neu_req_add_plugin {
@@ -196,6 +197,7 @@ typedef struct neu_resp_plugin_info {
neu_plugin_kind_e kind;
neu_node_type_e type;
bool single;
bool display;
char single_name[NEU_NODE_NAME_LEN];
} neu_resp_plugin_info_t;
include/neuron/define.h
-1
View File
@@ -51,7 +51,6 @@ typedef enum {
} neu_node_link_state_e;
typedef enum {
NEU_NODE_RUNNING_STATE_IDLE = 0,
NEU_NODE_RUNNING_STATE_INIT = 1,
NEU_NODE_RUNNING_STATE_READY = 2,
NEU_NODE_RUNNING_STATE_RUNNING = 3,
include/neuron/errcodes.h
+7 -6
View File
@@ -71,12 +71,13 @@ typedef enum {
NEU_ERR_TAG_DESCRIPTION_TOO_LONG = 2208,
NEU_ERR_TAG_PRECISION_INVALID = 2209,
NEU_ERR_LIBRARY_NOT_FOUND = 2301,
NEU_ERR_LIBRARY_INFO_INVALID = 2302,
NEU_ERR_LIBRARY_NAME_CONFLICT = 2303,
NEU_ERR_LIBRARY_FAILED_TO_OPEN = 2304,
NEU_ERR_LIBRARY_MODULE_INVALID = 2305,
NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL = 2306,
NEU_ERR_LIBRARY_NOT_FOUND = 2301,
NEU_ERR_LIBRARY_INFO_INVALID = 2302,
NEU_ERR_LIBRARY_NAME_CONFLICT = 2303,
NEU_ERR_LIBRARY_FAILED_TO_OPEN = 2304,
NEU_ERR_LIBRARY_MODULE_INVALID = 2305,
NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL = 2306,
NEU_ERR_LIBRARY_NOT_ALLOW_CREATE_INSTANCE = 2307,
NEU_ERR_LICENSE_NOT_FOUND = 2400,
NEU_ERR_LICENSE_INVALID = 2401,
persistence/0000_2.2.0_initial.sql
+1 -1
View File
@@ -25,7 +25,7 @@ CREATE TABLE IF NOT EXISTS
nodes (
name TEXT PRIMARY KEY check(length(name) <= 32),
type integer(1) NOT NULL check(type IN (1, 2)),
state integer(1) NOT NULL check(state BETWEEN 0 AND 4),
state integer(1) NOT NULL check(state BETWEEN 1 AND 4),
plugin_name TEXT NOT NULL check(length(plugin_name) <= 32)
);
plugins/restful/http.c
+1
View File
@@ -336,6 +336,7 @@ int http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_LIBRARY_NAME_CONFLICT:
case NEU_ERR_GROUP_EXIST:
case NEU_ERR_GROUP_NOT_ALLOW:
case NEU_ERR_LIBRARY_NOT_ALLOW_CREATE_INSTANCE:
status = NNG_HTTP_STATUS_CONFLICT;
break;
case NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT:
src/adapter/adapter.c
+7 -9
View File
@@ -54,7 +54,7 @@ static const adapter_callbacks_t callback_funs = {
.response = adapter_response,
};
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
{
int rv = 0;
neu_adapter_t * adapter = NULL;
@@ -119,10 +119,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
free(adapter->setting);
adapter->setting = NULL;
}
if (adapter->state == NEU_NODE_RUNNING_STATE_READY && start) {
adapter->module->intf_funs->start(adapter->plugin);
adapter->state = NEU_NODE_RUNNING_STATE_RUNNING;
}
}
if (info->module->type == NEU_NA_TYPE_DRIVER) {
@@ -147,14 +143,18 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
return adapter;
}
void neu_adapter_init(neu_adapter_t *adapter)
void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
{
neu_reqresp_head_t header = { .type = NEU_REQ_NODE_INIT };
neu_req_node_init_t init = { 0 };
nng_msg * init_msg = NULL;
strcpy(header.sender, adapter->name);
strcpy(header.receiver, "manager");
strcpy(init.node, adapter->name);
init_msg = neu_msg_gen(&header, &init);
init.auto_start = auto_start;
init_msg = neu_msg_gen(&header, &init);
nng_sendmsg(adapter->sock, init_msg, 0);
}
@@ -690,7 +690,6 @@ int neu_adapter_start(neu_adapter_t *adapter)
neu_err_code_e error = NEU_ERR_SUCCESS;
switch (adapter->state) {
case NEU_NODE_RUNNING_STATE_IDLE:
case NEU_NODE_RUNNING_STATE_INIT:
error = NEU_ERR_NODE_NOT_READY;
break;
@@ -730,7 +729,6 @@ int neu_adapter_stop(neu_adapter_t *adapter)
neu_err_code_e error = NEU_ERR_SUCCESS;
switch (adapter->state) {
case NEU_NODE_RUNNING_STATE_IDLE:
case NEU_NODE_RUNNING_STATE_INIT:
case NEU_NODE_RUNNING_STATE_READY:
error = NEU_ERR_NODE_NOT_RUNNING;
src/adapter/adapter_internal.h
+2 -2
View File
@@ -61,8 +61,8 @@ typedef struct adapter_msg_handler {
adapter_handler handler;
} adapter_msg_handler_t;
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start);
void neu_adapter_init(neu_adapter_t *adapter);
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
void neu_adapter_init(neu_adapter_t *adapter, bool auto_start);
int neu_adapter_start(neu_adapter_t *adapter);
int neu_adapter_start_single(neu_adapter_t *adapter);
src/core/manager.c
+11 -4
View File
@@ -223,6 +223,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
nng_pipe pipe = nng_msg_get_pipe(msg);
neu_node_manager_update(manager->node_manager, init->node, pipe);
if (init->auto_start) {
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, init->node);
neu_adapter_start(adapter);
}
nlog_notice("bind node %s to pipe(%d)", init->node, pipe.id);
break;
}
@@ -587,9 +594,9 @@ static void start_static_adapter(neu_manager_t *manager, const char *name)
adapter_info.handle = instance.handle;
adapter_info.module = instance.module;
adapter = neu_adapter_create(&adapter_info, manager);
adapter = neu_adapter_create(&adapter_info);
neu_node_manager_add_static(manager->node_manager, adapter);
neu_adapter_init(adapter);
neu_adapter_init(adapter, false);
neu_adapter_start(adapter);
}
@@ -610,10 +617,10 @@ static void start_single_adapter(neu_manager_t *manager, const char *name,
adapter_info.handle = instance.handle;
adapter_info.module = instance.module;
adapter = neu_adapter_create(&adapter_info, true);
adapter = neu_adapter_create(&adapter_info);
neu_node_manager_add_single(manager->node_manager, adapter, display);
neu_adapter_init(adapter);
neu_adapter_init(adapter, false);
neu_adapter_start_single(adapter);
}
src/core/manager_internal.c
+6 -2
View File
@@ -57,6 +57,10 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
return NEU_ERR_LIBRARY_NOT_FOUND;
}
if (info.single) {
return NEU_ERR_LIBRARY_NOT_ALLOW_CREATE_INSTANCE;
}
adapter = neu_node_manager_find(manager->node_manager, node_name);
if (adapter != NULL) {
return NEU_ERR_NODE_EXIST;
@@ -70,9 +74,9 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
adapter_info.handle = instance.handle;
adapter_info.module = instance.module;
adapter = neu_adapter_create(&adapter_info, start);
adapter = neu_adapter_create(&adapter_info);
neu_node_manager_add(manager->node_manager, adapter);
neu_adapter_init(adapter);
neu_adapter_init(adapter, start);
if (instance.module->type == NEU_NA_TYPE_APP) {
for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) {
src/core/plugin_manager.c
+10 -3
View File
@@ -201,6 +201,7 @@ UT_array *neu_plugin_manager_get_single(neu_plugin_manager_t *mgr)
};
info.display = el->display;
info.single = el->single;
strncpy(info.single_name, el->single_name,
sizeof(info.single_name));
@@ -224,9 +225,15 @@ int neu_plugin_manager_find(neu_plugin_manager_t *mgr, const char *plugin_name,
HASH_FIND_STR(mgr->plugins, plugin_name, plugin);
if (plugin != NULL) {
ret = 0;
info->type = plugin->type;
info->kind = plugin->kind;
ret = 0;
info->single = plugin->single;
info->display = plugin->display;
info->type = plugin->type;
info->kind = plugin->kind;
if (plugin->single_name != NULL) {
strcpy(info->single_name, plugin->single_name);
}
strncpy(info->name, plugin->name, sizeof(info->name));
strncpy(info->library, plugin->lib_name, sizeof(info->library));
strncpy(info->description, plugin->description,
src/core/storage.c
+3 -2
View File
@@ -120,8 +120,9 @@ int manager_load_node(neu_manager_t *manager)
manager, node_info->name, node_info->plugin_name,
node_info->state == NEU_NODE_RUNNING_STATE_RUNNING);
const char *ok_or_err = (0 == rv) ? "success" : "fail";
nlog_info("load adapter %s type:%d, name:%s plugin:%s", ok_or_err,
node_info->type, node_info->name, node_info->plugin_name);
nlog_info("load adapter %s type:%d, name:%s plugin:%s state:%d",
ok_or_err, node_info->type, node_info->name,
node_info->plugin_name, node_info->state);
}
utarray_free(node_infos);
src/persist/persist.c
+4 -12
View File
@@ -320,6 +320,8 @@ static inline int execute_sql(sqlite3 *db, const char *sql, ...)
if (SQLITE_OK != sqlite3_exec(db, query, NULL, NULL, &err_msg)) {
nlog_error("query `%s` fail: %s", query, err_msg);
rv = NEU_ERR_EINTERNAL;
} else {
nlog_info("query %s success", query);
}
sqlite3_free(err_msg);
@@ -561,18 +563,7 @@ neu_persister_t *neu_persister_create(const char *dir_name)
goto error;
}
int n = path_cat(path, dir_len, sizeof(path), "adapters");
if (sizeof(path) == n) {
nlog_error("path too long: %s", path);
goto error;
}
rv = create_dir(path);
if (rv != 0) {
nlog_error("failed to create directory: %s", path);
goto error;
}
n = path_cat(path, dir_len, sizeof(path), "plugins.json");
int n = path_cat(path, dir_len, sizeof(path), "plugins.json");
if (sizeof(path) == n) {
nlog_error("path too long: %s", path);
goto error;
@@ -607,6 +598,7 @@ neu_persister_t *neu_persister_create(const char *dir_name)
nlog_error("db `%s` fail: %s", path, sqlite3_errstr(rv));
goto error;
}
sqlite3_busy_timeout(persister->db, 100 * 1000);
// we rely on sqlite foreign key support
if (SQLITE_OK !=
sqlite3_exec(persister->db, "PRAGMA foreign_keys=ON", NULL, NULL,