diff --git a/ft/resource/api.resource b/ft/resource/api.resource index 8ff35e9b..106a9dd4 100644 --- a/ft/resource/api.resource +++ b/ft/resource/api.resource @@ -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 diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 15e0a6c3..bb31aa13 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -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 { diff --git a/include/neuron/define.h b/include/neuron/define.h index 475de606..8fae9378 100644 --- a/include/neuron/define.h +++ b/include/neuron/define.h @@ -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, diff --git a/persistence/0000_2.2.0_initial.sql b/persistence/0000_2.2.0_initial.sql index a5bab142..d95e78b7 100644 --- a/persistence/0000_2.2.0_initial.sql +++ b/persistence/0000_2.2.0_initial.sql @@ -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) ); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 05c2040a..4ba7dce9 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -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; diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 23ab7869..68595183 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -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); diff --git a/src/core/manager.c b/src/core/manager.c index c5c9eabb..8440d26b 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -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); } diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index bec39854..f90ffc49 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -70,9 +70,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++) { diff --git a/src/core/storage.c b/src/core/storage.c index f16c52d8..9fd860d9 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -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); diff --git a/src/persist/persist.c b/src/persist/persist.c index ee659a92..9b5b9460 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -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,