diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index 6f1b7440..67f1d6bd 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -142,6 +142,7 @@ typedef enum { NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE = 3020, NEU_ERR_PLUGIN_NOT_SUPPORT_SCAN_TAGS = 3021, NEU_ERR_PLUGIN_NOT_SUPPORT_TEST_READ_TAG = 3022, + NEU_ERR_PLUGIN_NOT_SUPPORT_CMD_CALL = 3023, NEU_ERR_MQTT_FAILURE = 4000, NEU_ERR_MQTT_NO_CERTFILESET = 4001, diff --git a/include/neuron/json/neu_json_driver.h b/include/neuron/json/neu_json_driver.h new file mode 100644 index 00000000..a26468e9 --- /dev/null +++ b/include/neuron/json/neu_json_driver.h @@ -0,0 +1,41 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef _NEU_JSON_API_DRIVER_CMD_H_ +#define _NEU_JSON_API_DRIVER_CMD_H_ + +#include "define.h" + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct { + char *driver; + char *cmd; +} neu_json_driver_cmd_t; + +int neu_json_decode_driver_cmd_req(char *buf, neu_json_driver_cmd_t **result); +void neu_json_decode_driver_cmd_req_free(neu_json_driver_cmd_t *req); + +#ifdef __cplusplus +} +#endif + +#endif \ No newline at end of file diff --git a/include/neuron/msg.h b/include/neuron/msg.h index b203c9dc..f4613f1b 100644 --- a/include/neuron/msg.h +++ b/include/neuron/msg.h @@ -125,6 +125,8 @@ typedef enum neu_reqresp_type { NEU_REQ_CHECK_SCHEMA, NEU_RESP_CHECK_SCHEMA, + + NEU_REQ_DRIVER_CMD, } neu_reqresp_type_e; static const char *neu_reqresp_type_string_t[] = { @@ -210,6 +212,8 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_CHECK_SCHEMA] = "NEU_REQ_CHECK_SCHEMA", [NEU_RESP_CHECK_SCHEMA] = "NEU_RESP_CHECK_SCHEMA", + + [NEU_REQ_DRIVER_CMD] = "NEU_REQ_DRIVER_CMD", }; inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type) @@ -238,6 +242,11 @@ typedef struct neu_resp_error { int error; } neu_resp_error_t; +typedef struct neu_req_driver_cmd { + char driver[NEU_NODE_NAME_LEN]; + char *cmd; +} neu_req_driver_cmd_t; + typedef struct neu_req_check_schema { char schema[NEU_PLUGIN_NAME_LEN]; } neu_req_check_schema_t; diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 85ff7ebf..712c0af6 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -114,6 +114,7 @@ typedef struct neu_plugin_intf_funs { char *ctx); int (*test_read_tag)(neu_plugin_t *plugin, void *req, neu_datatag_t tag); + int (*call)(neu_plugin_t *plugin, const char *cmd); } driver; }; diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 624f71ee..b0bcf1f5 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -22,6 +22,7 @@ #include "otel/otel_manager.h" #include "utils/asprintf.h" #include "version.h" +#include "json/neu_json_driver.h" #include "json/neu_json_mqtt.h" #include "json/neu_json_rw.h" @@ -179,6 +180,26 @@ static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states, return json_str; } +static inline int send_driver_cmd(neu_plugin_t * plugin, + neu_json_driver_cmd_t *req) +{ + plog_notice(plugin, "driver cmd, driver:%s, cmd:%s", req->driver, req->cmd); + neu_reqresp_head_t header = { 0 }; + header.ctx = NULL; + + neu_req_driver_cmd_t cmd = { 0 }; + strncpy(cmd.driver, req->driver, NEU_NODE_NAME_LEN); + cmd.cmd = strdup(req->cmd); + + if (0 != neu_plugin_op(plugin, header, &cmd)) { + free(cmd.cmd); + plog_error(plugin, "neu_plugin_op(NEU_REQ_DRIVER_CMD) fail"); + return -1; + } + + return 0; +} + static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_read_req_t *req) { @@ -533,6 +554,30 @@ end: return rv; } +void handle_driver_cmd_req(neu_mqtt_qos_e qos, const char *topic, + const uint8_t *payload, uint32_t len, void *data, + trace_w3c_t *trace_w3c) +{ + (void) qos; + (void) topic; + (void) trace_w3c; + neu_plugin_t * plugin = data; + neu_json_driver_cmd_t *req = NULL; + + char *json_str = calloc(len + 1, sizeof(char)); + memcpy(json_str, payload, len); + + int rv = neu_json_decode_driver_cmd_req(json_str, &req); + if (rv != 0) { + plog_error(plugin, "neu_json_decode_driver_cmd_req failed"); + free(json_str); + return; + } + + neu_json_decode_driver_cmd_req_free(req); + free(json_str); +} + void handle_read_req(neu_mqtt_qos_e qos, const char *topic, const uint8_t *payload, uint32_t len, void *data, trace_w3c_t *trace_w3c) diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index 121e3b3c..9ca6c980 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/plugins/mqtt/mqtt_handle.h @@ -50,6 +50,10 @@ void handle_read_req(neu_mqtt_qos_e qos, const char *topic, int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json, neu_resp_read_group_t *data); +void handle_driver_cmd_req(neu_mqtt_qos_e qos, const char *topic, + const uint8_t *payload, uint32_t len, void *data, + trace_w3c_t *trace_w3c); + char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data, mqtt_upload_format_e format); int handle_trans_data(neu_plugin_t * plugin, diff --git a/plugins/mqtt/mqtt_plugin.h b/plugins/mqtt/mqtt_plugin.h index 94d98c94..120e19e8 100644 --- a/plugins/mqtt/mqtt_plugin.h +++ b/plugins/mqtt/mqtt_plugin.h @@ -52,6 +52,7 @@ struct neu_plugin { char * read_req_topic; char * read_resp_topic; char * upload_topic; + char * driver_cmd_topic; route_entry_t * route_tbl; int (*parse_config)(neu_plugin_t *plugin, const char *setting, diff --git a/plugins/mqtt/mqtt_plugin_intf.c b/plugins/mqtt/mqtt_plugin_intf.c index f08621d9..415ac337 100644 --- a/plugins/mqtt/mqtt_plugin_intf.c +++ b/plugins/mqtt/mqtt_plugin_intf.c @@ -178,6 +178,8 @@ int mqtt_plugin_uninit(neu_plugin_t *plugin) plugin->read_resp_topic = NULL; free(plugin->upload_topic); plugin->upload_topic = NULL; + free(plugin->driver_cmd_topic); + plugin->driver_cmd_topic = NULL; route_tbl_free(plugin->route_tbl); @@ -279,6 +281,14 @@ static int create_topic(neu_plugin_t *plugin) return -1; } + neu_asprintf(&plugin->driver_cmd_topic, "/neuron/%s/driver/cmd", + plugin->common.name); + if (NULL == plugin->driver_cmd_topic) { + free(plugin->driver_cmd_topic); + plugin->driver_cmd_topic = NULL; + return -1; + } + return 0; } @@ -306,6 +316,14 @@ static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config) return NEU_ERR_MQTT_SUBSCRIBE_FAILURE; } + if (0 != + neu_mqtt_client_subscribe(plugin->client, config->qos, + plugin->driver_cmd_topic, plugin, + handle_driver_cmd_req)) { + plog_error(plugin, "subscribe [%s] fail", plugin->driver_cmd_topic); + return NEU_ERR_MQTT_SUBSCRIBE_FAILURE; + } + return 0; } @@ -313,6 +331,7 @@ static int unsubscribe(neu_plugin_t *plugin, const mqtt_config_t *config) { neu_mqtt_client_unsubscribe(plugin->client, plugin->read_req_topic); neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic); + neu_mqtt_client_unsubscribe(plugin->client, plugin->driver_cmd_topic); neu_msleep(100); // wait for message completion return 0; } diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 169f7357..494975f8 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -443,6 +443,11 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, strcpy(pheader->sender, adapter->name); switch (pheader->type) { + case NEU_REQ_DRIVER_CMD: { + neu_req_driver_cmd_t *cmd = (neu_req_driver_cmd_t *) data; + strcpy(pheader->receiver, cmd->driver); + break; + } case NEU_REQ_READ_GROUP: { neu_req_read_group_t *cmd = (neu_req_read_group_t *) data; strcpy(pheader->receiver, cmd->driver); @@ -1399,6 +1404,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_msg_free(msg); break; } + case NEU_REQ_DRIVER_CMD: { + neu_req_driver_cmd_t *cmd = (neu_req_driver_cmd_t *) &header[1]; + neu_adapter_driver_cmd((neu_adapter_driver_t *) adapter, + (const char *) cmd->cmd); + free(cmd->cmd); + break; + } default: nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name, neu_reqresp_type_string(header->type)); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 5d464f8c..900dfe43 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -3021,4 +3021,16 @@ void neu_adapter_driver_test_read_tag(neu_adapter_driver_t *driver, if (g != NULL && neu_group_tag_size(g->group) > 0) { start_group_timer(driver, g); } +} + +int neu_adapter_driver_cmd(neu_adapter_driver_t *driver, const char *cmd) +{ + if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { + return NEU_ERR_PLUGIN_NOT_RUNNING; + } + if (driver->adapter.module->intf_funs->driver.call == NULL) { + return NEU_ERR_PLUGIN_NOT_SUPPORT_CMD_CALL; + } + return driver->adapter.module->intf_funs->driver.call( + driver->adapter.plugin, cmd); } \ No newline at end of file diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index 352852df..74e7837a 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -94,4 +94,7 @@ void neu_adapter_driver_unsubscribe(neu_adapter_driver_t * driver, void neu_adapter_driver_scan_tags(neu_adapter_driver_t *driver, neu_reqresp_head_t * req); + +int neu_adapter_driver_cmd(neu_adapter_driver_t *driver, const char *cmd); + #endif diff --git a/src/base/msg_internal.h b/src/base/msg_internal.h index 3c6df2c2..62fe127f 100644 --- a/src/base/msg_internal.h +++ b/src/base/msg_internal.h @@ -101,7 +101,8 @@ extern "C" { XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \ XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t) \ XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t) \ - XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t) + XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t) \ + XX(NEU_REQ_DRIVER_CMD, neu_req_driver_cmd_t) static inline size_t neu_reqresp_size(neu_reqresp_type_e t) { diff --git a/src/core/manager.c b/src/core/manager.c index b2d9d8a9..c50da527 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -1062,6 +1062,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_WRITE_GTAGS: case NEU_REQ_GET_TAG: case NEU_REQ_NODE_CTL: + case NEU_REQ_DRIVER_CMD: case NEU_REQ_ADD_GROUP: { neu_otel_trace_ctx trace = NULL; neu_otel_scope_ctx scope = NULL; diff --git a/src/parser/neu_json_driver.c b/src/parser/neu_json_driver.c new file mode 100644 index 00000000..21d9ea04 --- /dev/null +++ b/src/parser/neu_json_driver.c @@ -0,0 +1,90 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ +#include +#include + +#include + +#include "json/json.h" + +#include "json/neu_json_driver.h" + +int neu_json_decode_driver_cmd_req_json(void * json_obj, + neu_json_driver_cmd_t *req) +{ + int ret = 0; + + neu_json_elem_t req_elems[] = { + { + .name = "driver", + .t = NEU_JSON_STR, + }, + { + .name = "cmd", + .t = NEU_JSON_STR, + }, + }; + ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems), + req_elems); + if (ret != 0) { + return -1; + } + + req->driver = req_elems[0].v.val_str; + req->cmd = req_elems[1].v.val_str; + + return 0; +} + +int neu_json_decode_driver_cmd_req(char *buf, neu_json_driver_cmd_t **result) +{ + int ret = 0; + void * json_obj = NULL; + neu_json_driver_cmd_t *req = calloc(1, sizeof(neu_json_driver_cmd_t)); + if (req == NULL) { + return -1; + } + + json_obj = neu_json_decode_new(buf); + if (NULL == json_obj) { + free(req); + return -1; + } + + ret = neu_json_decode_driver_cmd_req_json(json_obj, req); + if (0 == ret) { + *result = req; + } else { + free(req); + } + + neu_json_decode_free(json_obj); + return ret; + + return 0; +} + +void neu_json_decode_driver_cmd_req_free(neu_json_driver_cmd_t *req) +{ + if (req) { + free(req->driver); + free(req->cmd); + free(req); + } +} \ No newline at end of file diff --git a/src/utils/json.c b/src/utils/json.c index 8cd6890f..819a3e51 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -319,7 +319,7 @@ static int decode_object(json_t *root, neu_json_elem_t *ele) case NEU_JSON_INT: ele->v.val_int = json_integer_value(ob); break; - case NEU_JSON_STR: + case NEU_JSON_STR: { if (!json_is_string(ob)) { zlog_error(neuron, "json decode: %s failed", ele->name); return -1; @@ -331,6 +331,7 @@ static int decode_object(json_t *root, neu_json_elem_t *ele) } ele->v.val_str = strdup(str_val); break; + } case NEU_JSON_FLOAT: if (json_is_integer(ob)) { ele->v.val_float = (float) json_integer_value(ob);