Add node state: running state and link state & api: start/stop node

note start/stop node operation

include/neu_adapter.h

@@ -75,6 +75,9 @@ typedef enum neu_reqresp_type {
     NEU_REQRESP_SET_NODE_SETTING,
     NEU_REQRESP_GET_NODE_SETTING,
     NEU_REQRESP_GET_NODE_SETTING_RESP,
+    NEU_REQRESP_NODE_STATE,
+    NEU_REQRESP_NODE_STATE_RESP,
+    NEU_REQRESP_NODE_CTL,
 } neu_reqresp_type_e;
 
 typedef uint32_t neu_node_id_t;
@@ -279,6 +282,27 @@ typedef struct neu_reqresp_node_setting {
     char *  setting;
 } neu_reqresp_node_setting_t;
 
+/* NEU_REQRESP_GET_NODE_STATE */
+typedef struct neu_cmd_get_node_state {
+    neu_node_id_t node_id;
+} neu_cmd_get_node_state_t;
+
+typedef struct neu_reqresp_node_state {
+    int32_t            result;
+    neu_plugin_state_t state;
+} neu_reqresp_node_state_t;
+
+typedef enum neu_adapter_ctl {
+    NEU_ADAPTER_CTL_START = 0,
+    NEU_ADAPTER_CTL_STOP,
+} neu_adapter_ctl_e;
+
+/* NEU_REQRESP_NODE_CTL */
+typedef struct neu_cmd_node_ctl {
+    neu_node_id_t     node_id;
+    neu_adapter_ctl_e ctl;
+} neu_cmd_node_ctl_t;
+
 typedef struct neu_request {
     uint32_t           req_id;
     neu_reqresp_type_e req_type;

include/neu_plugin.h

@@ -29,16 +29,9 @@ extern "C" {
 #define NEURON_PLUGIN_VER_1_0 100
 #define NEURON_PLUGIN_VER_2_0 200
 
-typedef enum neu_plugin_state {
-    NEURON_PLUGIN_STATE_NULL,
-    NEURON_PLUGIN_STATE_IDLE,
-    NEURON_PLUGIN_STATE_READY,
-    NEURON_PLUGIN_STATE_RUNNING,
-} neu_plugin_state_e;
-
 typedef struct neu_plugin_common {
     uint32_t                   magic;
-    neu_plugin_state_e         state;
+    neu_plugin_state_t         state;
     neu_adapter_t *            adapter;
     const adapter_callbacks_t *adapter_callbacks;
     uint32_t                   event_id;
@@ -138,6 +131,10 @@ intptr_t neu_plugin_set_node_setting(neu_plugin_t *plugin,
                                      const char *  setting);
 int32_t neu_plugin_get_node_setting(neu_plugin_t *plugin, neu_node_id_t node_id,
                                     char **setting);
+int32_t neu_plugin_get_node_state(neu_plugin_t *plugin, neu_node_id_t node_id,
+                                  neu_plugin_state_t *state);
+intptr_t neu_plugin_node_ctl(neu_plugin_t *plugin, neu_node_id_t node_id,
+                             neu_adapter_ctl_e ctl);
 #ifdef __cplusplus
 }
 #endif

include/neu_plugin_info.h

@@ -34,6 +34,24 @@ typedef enum plugin_kind {
     PLUGIN_KIND_CUSTOM
 } plugin_kind_e;
 
+typedef enum neu_plugin_link_state {
+    NEU_PLUGIN_LINK_STATE_DISCONNECTED = 0,
+    NEU_PLUGIN_LINK_STATE_CONNECTING,
+    NEU_PLUGIN_LINK_STATE_CONNECTED,
+} neu_plugin_link_state_e;
+
+typedef enum neu_plugin_running_state {
+    NEU_PLUGIN_RUNNING_STATE_IDLE = 0,
+    NEU_PLUGIN_RUNNING_STATE_READY,
+    NEU_PLUGIN_RUNNING_STATE_RUNNING,
+    NEU_PLUGIN_RUNNING_STATE_STOPPED,
+} neu_plugin_running_state_e;
+
+typedef struct neu_plugin_state {
+    neu_plugin_link_state_e    link;
+    neu_plugin_running_state_e running;
+} neu_plugin_state_t;
+
 #ifdef __cplusplus
 }
 #endif

plugins/modbus/modbus_tcp.c

@@ -482,7 +482,9 @@ static neu_plugin_t *modbus_tcp_open(neu_adapter_t *            adapter,
 
     plugin->common.adapter           = adapter;
     plugin->common.adapter_callbacks = callbacks;
-    plugin->common.state             = NEURON_PLUGIN_STATE_NULL;
+    //    plugin->common.state             = NEURON_PLUGIN_STATE_NULL;
+    plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_IDLE;
+    plugin->common.state.link    = NEU_PLUGIN_LINK_STATE_CONNECTING;
     TAILQ_INIT(&plugin->sub_instances);
 
     return plugin;
@@ -499,8 +501,10 @@ static int modbus_tcp_init(neu_plugin_t *plugin)
     pthread_mutex_init(&plugin->mtx, NULL);
     plugin->client = neu_tcp_client_create("192.168.50.68", 502);
 
-    plugin->common.state = NEURON_PLUGIN_STATE_READY;
+    //    plugin->common.state = NEURON_PLUGIN_STATE_READY;
     log_info("modbus tcp init.....");
+    plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_RUNNING;
+
     return 0;
 }
 
@@ -529,6 +533,7 @@ static int modbus_tcp_uninit(neu_plugin_t *plugin)
     pthread_mutex_destroy(&plugin->mtx);
 
     log_info("modbus uninit end...");
+    plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_STOPPED;
 
     return 0;
 }
@@ -542,6 +547,7 @@ static int modbus_tcp_config(neu_plugin_t *plugin, neu_config_t *configs)
     log_info("modbus config............. type: %d, config: %s", configs->type,
              configs->buf);
 
+    // plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_READY;
     return ret;
 }
 

plugins/restful/adapter_handle.c

@@ -166,4 +166,52 @@ void handle_get_node_setting(nng_aio *aio)
 
     http_ok(aio, setting);
     free(setting);
+}
+
+void handle_node_ctl(nng_aio *aio)
+{
+    neu_plugin_t *plugin = neu_rest_get_plugin();
+
+    REST_PROCESS_HTTP_REQUEST(
+        aio, neu_parse_node_ctl_req_t, neu_parse_decode_node_ctl, {
+            intptr_t err = 0;
+
+            err = neu_plugin_node_ctl(plugin, req->node_id, req->cmd);
+
+            if (err != 0) {
+                http_bad_request(aio, "{\"error\": 1}");
+            } else {
+                http_ok(aio, "{\"error\": 0}");
+            }
+        })
+}
+
+void handle_get_node_state(nng_aio *aio)
+{
+    neu_plugin_t *             plugin    = neu_rest_get_plugin();
+    char *                     s_node_id = http_get_param(aio, "node_id");
+    neu_node_id_t              node_id   = 0;
+    neu_parse_node_state_res_t res       = { 0 };
+    neu_plugin_state_t         state     = { 0 };
+    char *                     result    = NULL;
+
+    if (s_node_id == NULL) {
+        http_bad_request(aio, "{\"error\": 1}");
+        return;
+    }
+
+    node_id = (neu_node_id_t) atoi(s_node_id);
+
+    if (neu_plugin_get_node_state(plugin, node_id, &state) != 0) {
+        http_not_found(aio, "{\"error\": 1}");
+        return;
+    }
+
+    res.running = state.running;
+    res.link    = state.link;
+
+    neu_json_encode_by_fn(&res, neu_parse_encode_get_node_state, &result);
+
+    http_ok(aio, result);
+    free(result);
 }

plugins/restful/adapter_handle.h

@@ -27,5 +27,7 @@ void handle_update_adapter(nng_aio *aio);
 void handle_get_adapter(nng_aio *aio);
 void handle_set_node_setting(nng_aio *aio);
 void handle_get_node_setting(nng_aio *aio);
+void handle_node_ctl(nng_aio *aio);
+void handle_get_node_state(nng_aio *aio);
 
 #endif

plugins/restful/handle.c

@@ -95,6 +95,12 @@ struct neu_rest_handler cors_handler[] = {
     {
         .url = "/api/v2/node/setting",
     },
+    {
+        .url = "/api/v2/node/ctl",
+    },
+    {
+        .url = "/api/v2/node/state",
+    },
 };
 
 struct neu_rest_handler api_handlers[] = {
@@ -268,6 +274,18 @@ struct neu_rest_handler api_handlers[] = {
         .url           = "/api/v2/node/setting",
         .value.handler = handle_get_node_setting,
     },
+    {
+        .method        = NEU_REST_METHOD_POST,
+        .type          = NEU_REST_HANDLER_FUNCTION,
+        .url           = "/api/v2/node/ctl",
+        .value.handler = handle_node_ctl,
+    },
+    {
+        .method        = NEU_REST_METHOD_GET,
+        .type          = NEU_REST_HANDLER_FUNCTION,
+        .url           = "/api/v2/node/state",
+        .value.handler = handle_get_node_state,
+    },
 };
 
 void neu_rest_web_handler(const struct neu_rest_handler **handlers,

src/adapter/adapter.c

@@ -104,11 +104,10 @@
     }
 
 typedef enum adapter_state {
-    ADAPTER_STATE_NULL,
-    ADAPTER_STATE_IDLE,
-    ADAPTER_STATE_STARTING,
+    ADAPTER_STATE_IDLE = 0,
+    ADAPTER_STATE_READY,
     ADAPTER_STATE_RUNNING,
-    ADAPTER_STATE_STOPPING,
+    ADAPTER_STATE_STOPPED,
 } adapter_state_e;
 
 struct neu_adapter {
@@ -149,11 +148,12 @@ static void adapter_loop(void *arg)
 {
     int            rv;
     neu_adapter_t *adapter;
+    nng_dialer     dialer = { 0 };
 
     adapter = (neu_adapter_t *) arg;
     const char *manager_url;
     manager_url = neu_manager_get_url(adapter->manager);
-    rv          = nng_dial(adapter->sock, manager_url, NULL, 0);
+    rv          = nng_dial(adapter->sock, manager_url, &dialer, 0);
     if (rv != 0) {
         neu_panic("The adapter can't dial to %s", manager_url);
     }
@@ -183,7 +183,7 @@ static void adapter_loop(void *arg)
 
         nng_mtx_lock(adapter->mtx);
         if (adapter->stop) {
-            adapter->state = ADAPTER_STATE_NULL;
+            adapter->state = ADAPTER_STATE_STOPPED;
             nng_mtx_unlock(adapter->mtx);
             log_info("Exit loop of the adapter(%s)", adapter->name);
             break;
@@ -396,7 +396,7 @@ static void adapter_loop(void *arg)
             log_info("adapter(%s) exit loop by exit_code=%d", adapter->name,
                      exit_code);
             nng_mtx_lock(adapter->mtx);
-            adapter->state = ADAPTER_STATE_NULL;
+            adapter->state = ADAPTER_STATE_STOPPED;
             adapter->stop  = true;
             nng_mtx_unlock(adapter->mtx);
             break;
@@ -411,6 +411,7 @@ static void adapter_loop(void *arg)
         nng_msg_free(msg);
     }
 
+    nng_dialer_close(dialer);
     return;
 }
 
@@ -729,6 +730,28 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
         break;
     }
 
+    case NEU_REQRESP_NODE_STATE: {
+        ADAPTER_RESP_CMD(
+            adapter, cmd, neu_reqresp_node_state_t, neu_cmd_get_node_state_t,
+            rv, NEU_REQRESP_NODE_STATE_RESP, p_result, {
+                ret = calloc(1, sizeof(neu_reqresp_node_state_t));
+
+                ret->result = neu_manager_adapter_get_state(
+                    adapter->manager, req_cmd->node_id, &ret->state);
+            });
+        break;
+    }
+
+    case NEU_REQRESP_NODE_CTL: {
+        ADAPTER_RESP_CODE(adapter, cmd, intptr_t, neu_cmd_node_ctl_t, rv,
+                          NEU_REQRESP_ERR_CODE, p_result, {
+                              ret = neu_manager_adapter_ctl(adapter->manager,
+                                                            req_cmd->node_id,
+                                                            req_cmd->ctl);
+                          });
+        break;
+    }
+
     default:
         rv = -1;
         break;
@@ -896,7 +919,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info,
     }
 
     int rv;
-    adapter->state = ADAPTER_STATE_NULL;
+    adapter->state = ADAPTER_STATE_IDLE;
     adapter->stop  = false;
     if ((rv = nng_mtx_alloc(&adapter->mtx)) != 0) {
         log_error("Can't allocate mutex for adapter");
@@ -994,6 +1017,7 @@ int neu_adapter_start(neu_adapter_t *adapter)
         intf_funs->init(adapter->plugin);
     }
 
+    adapter->stop = false;
     nng_thread_create(&adapter->thrd, adapter_loop, adapter);
     return rv;
 }
@@ -1120,4 +1144,11 @@ int neu_adapter_get_setting(neu_adapter_t *adapter, char **config)
     }
 
     return -1;
+}
+
+neu_plugin_state_t neu_adapter_get_state(neu_adapter_t *adapter)
+{
+    neu_plugin_common_t *common = neu_plugin_to_plugin_common(adapter->plugin);
+
+    return common->state;
 }

src/adapter/adapter_internal.h

@@ -40,5 +40,6 @@ adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter);
 plugin_id_t    neu_adapter_get_plugin_id(neu_adapter_t *adapter);
 int neu_adapter_set_setting(neu_adapter_t *adapter, neu_config_t *config);
 int neu_adapter_get_setting(neu_adapter_t *adapter, char **config);
+neu_plugin_state_t neu_adapter_get_state(neu_adapter_t *adapter);
 
 #endif

src/base/neu_plugin_common.c

@@ -452,4 +452,34 @@ int32_t neu_plugin_get_node_setting(neu_plugin_t *plugin, neu_node_id_t node_id,
                         *setting = resp->setting;
                     })
     return ret;
+}
+
+int32_t neu_plugin_get_node_state(neu_plugin_t *plugin, neu_node_id_t node_id,
+                                  neu_plugin_state_t *state)
+{
+    neu_cmd_get_node_state_t get_node_state = { 0 };
+    int                      ret            = -1;
+
+    get_node_state.node_id = node_id;
+
+    PLUGIN_CALL_CMD(plugin, NEU_REQRESP_NODE_STATE, get_node_state,
+                    neu_reqresp_node_state_t, {
+                        ret    = resp->result;
+                        *state = resp->state;
+                    })
+    return ret;
+}
+
+intptr_t neu_plugin_node_ctl(neu_plugin_t *plugin, neu_node_id_t node_id,
+                             neu_adapter_ctl_e ctl)
+{
+    intptr_t           errorcode = -1;
+    neu_cmd_node_ctl_t node_ctl  = { 0 };
+
+    node_ctl.node_id = node_id;
+    node_ctl.ctl     = ctl;
+
+    PLUGIN_CALL_CMD(plugin, NEU_REQRESP_NODE_CTL, node_ctl, intptr_t,
+                    { errorcode = (intptr_t) resp; })
+    return errorcode;
 }

src/core/neu_manager.c

@@ -578,7 +578,7 @@ static int manager_unreg_adapter(neu_manager_t *manager, adapter_id_t id,
     return rv;
 }
 
-static int manager_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter)
+int neu_manager_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter)
 {
     int rv = 0;
 
@@ -602,7 +602,7 @@ static int manager_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter)
     return rv;
 }
 
-static int manager_stop_adapter(neu_manager_t *manager, neu_adapter_t *adapter)
+int neu_manager_stop_adapter(neu_manager_t *manager, neu_adapter_t *adapter)
 {
     int      rv = 0;
     size_t   msg_size;
@@ -681,7 +681,7 @@ static void reg_and_start_default_adapters(neu_manager_t *manager)
             manager, (adapter_reg_cmd_t *) &default_adapter_reg_cmds[i],
             &p_adapter);
         if (id != 0 && p_adapter != NULL) {
-            manager_start_adapter(manager, p_adapter);
+            neu_manager_start_adapter(manager, p_adapter);
         }
     }
     return;
@@ -701,7 +701,7 @@ static void stop_and_unreg_bind_adapters(neu_manager_t *manager)
         reg_entity = (adapter_reg_entity_t *) iterator_get(&iter);
         adapter    = reg_entity->adapter;
         if (adapter != NULL) {
-            manager_stop_adapter(manager, adapter);
+            neu_manager_stop_adapter(manager, adapter);
             manager_unreg_adapter(manager, reg_entity->adapter_id, false);
         }
     }
@@ -1828,5 +1828,85 @@ int neu_manager_adapter_get_setting(neu_manager_t *manager,
     ret = neu_adapter_get_setting(reg_entity->adapter, config);
     nng_mtx_unlock(manager->adapters_mtx);
 
+    return ret;
+}
+
+int neu_manager_adapter_get_state(neu_manager_t *manager, neu_node_id_t node_id,
+                                  neu_plugin_state_t *state)
+{
+    adapter_id_t          adapter_id = { 0 };
+    adapter_reg_entity_t *reg_entity = NULL;
+
+    nng_mtx_lock(manager->adapters_mtx);
+
+    adapter_id = neu_manager_adapter_id_from_node_id(manager, node_id);
+    reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id);
+
+    if (reg_entity == NULL) {
+        log_error("Can't find matched src registered adapter");
+        nng_mtx_unlock(manager->adapters_mtx);
+        return -1;
+    }
+
+    *state = neu_adapter_get_state(reg_entity->adapter);
+    nng_mtx_unlock(manager->adapters_mtx);
+
+    return 0;
+}
+
+int neu_manager_adapter_ctl(neu_manager_t *manager, neu_node_id_t node_id,
+                            neu_adapter_ctl_e ctl)
+{
+    adapter_id_t          adapter_id = { 0 };
+    adapter_reg_entity_t *reg_entity = NULL;
+    int                   ret        = -1;
+    neu_plugin_state_t    state      = { 0 };
+
+    nng_mtx_lock(manager->adapters_mtx);
+
+    adapter_id = neu_manager_adapter_id_from_node_id(manager, node_id);
+    reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id);
+
+    if (reg_entity == NULL) {
+        log_error("Can't find matched src registered adapter");
+        nng_mtx_unlock(manager->adapters_mtx);
+        return -1;
+    }
+
+    nng_mtx_unlock(manager->adapters_mtx);
+    state = neu_adapter_get_state(reg_entity->adapter);
+
+    switch (ctl) {
+    case NEU_ADAPTER_CTL_START:
+        switch (state.running) {
+        case NEU_PLUGIN_RUNNING_STATE_IDLE:
+            ret = -1;
+            break;
+        case NEU_PLUGIN_RUNNING_STATE_READY:
+        case NEU_PLUGIN_RUNNING_STATE_STOPPED:
+            // ret = neu_manager_start_adapter(manager, reg_entity->adapter);
+            ret = 0;
+            break;
+        case NEU_PLUGIN_RUNNING_STATE_RUNNING:
+            ret = 0;
+            break;
+        }
+        break;
+    case NEU_ADAPTER_CTL_STOP:
+        switch (state.running) {
+        case NEU_PLUGIN_RUNNING_STATE_IDLE:
+        case NEU_PLUGIN_RUNNING_STATE_READY:
+            ret = -1;
+            break;
+        case NEU_PLUGIN_RUNNING_STATE_RUNNING:
+            // ret = neu_manager_stop_adapter(manager, reg_entity->adapter);
+            ret = 0;
+            break;
+        case NEU_PLUGIN_RUNNING_STATE_STOPPED:
+            ret = 0;
+            break;
+        }
+        break;
+    }
     return ret;
 }

src/core/neu_manager.h

@@ -68,4 +68,10 @@ int                  neu_manager_adapter_set_setting(neu_manager_t *manager,
                                                      neu_node_id_t node_id, const char *config);
 int                  neu_manager_adapter_get_setting(neu_manager_t *manager,
                                                      neu_node_id_t node_id, char **config);
+int neu_manager_adapter_get_state(neu_manager_t *manager, neu_node_id_t node_id,
+                                  neu_plugin_state_t *state);
+int neu_manager_adapter_ctl(neu_manager_t *manager, neu_node_id_t node_id,
+                            neu_adapter_ctl_e ctl);
+int neu_manager_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter);
+int neu_manager_stop_adapter(neu_manager_t *manager, neu_adapter_t *adapter);
 #endif

src/parser/neu_json_node.c

@@ -210,6 +210,58 @@ int neu_parse_decode_node_setting(char *                         buf,
 }
 
 void neu_parse_decode_node_setting_free(neu_parse_node_setting_req_t *req)
+{
+    free(req);
+}
+
+int neu_parse_encode_get_node_state(void *json_object, void *param)
+{
+    neu_parse_node_state_res_t *res = (neu_parse_node_state_res_t *) param;
+
+    neu_json_elem_t elems[] = {
+        {
+            .name      = "running",
+            .t         = NEU_JSON_INT,
+            .v.val_int = res->running,
+        },
+        {
+            .name      = "link",
+            .t         = NEU_JSON_INT,
+            .v.val_int = res->link,
+        },
+    };
+
+    return neu_json_encode_field(json_object, elems, NEU_JSON_ELEM_SIZE(elems));
+}
+
+int neu_parse_decode_node_ctl(char *buf, neu_parse_node_ctl_req_t **result)
+{
+    neu_parse_node_ctl_req_t *req = calloc(1, sizeof(neu_parse_node_ctl_req_t));
+    neu_json_elem_t           elem[] = {
+        {
+            .name = "id",
+            .t    = NEU_JSON_INT,
+        },
+        {
+            .name = "cmd",
+            .t    = NEU_JSON_INT,
+        },
+    };
+
+    int ret = neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elem), elem);
+    if (ret != 0) {
+        free(req);
+        return -1;
+    }
+    req->node_id = elem[0].v.val_int;
+    req->cmd     = elem[1].v.val_int;
+
+    *result = req;
+
+    return 0;
+}
+
+void neu_parse_decode_node_ctl_free(neu_parse_node_ctl_req_t *req)
 {
     free(req);
 }

src/parser/neu_json_node.h

@@ -80,6 +80,19 @@ int  neu_parse_decode_node_setting(char *                         buf,
                                    neu_parse_node_setting_req_t **result);
 void neu_parse_decode_node_setting_free(neu_parse_node_setting_req_t *req);
 
+typedef struct {
+    uint8_t running;
+    uint8_t link;
+} neu_parse_node_state_res_t;
+int neu_parse_encode_get_node_state(void *json_object, void *param);
+
+typedef struct {
+    uint32_t node_id;
+    uint8_t  cmd;
+} neu_parse_node_ctl_req_t;
+int  neu_parse_decode_node_ctl(char *buf, neu_parse_node_ctl_req_t **result);
+void neu_parse_decode_node_ctl_free(neu_parse_node_ctl_req_t *req);
+
 #ifdef __cplusplus
 }
 #endif