diff --git a/plugins/nano-mqtt/mqtt_handle.c b/plugins/nano-mqtt/mqtt_handle.c index 67424fd4..1571a7bb 100644 --- a/plugins/nano-mqtt/mqtt_handle.c +++ b/plugins/nano-mqtt/mqtt_handle.c @@ -17,8 +17,11 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ +#include "connection/mqtt_client.h" #include "errcodes.h" #include "version.h" +#include "json/neu_json_mqtt.h" +#include "json/neu_json_rw.h" #include "mqtt_handle.h" #include "mqtt_plugin.h" @@ -176,6 +179,49 @@ static char *generate_upload_json(neu_plugin_t * plugin, return json_str; } +static char *generate_read_resp_json(neu_plugin_t * plugin, + neu_json_mqtt_t * mqtt, + neu_resp_read_group_t *data) +{ + neu_resp_tag_value_t *tags = data->tags; + uint16_t len = data->n_tag; + char * json_str = NULL; + neu_json_read_resp_t json = { 0 }; + + if (0 != tag_values_to_json(tags, len, &json)) { + plog_error(plugin, "tag_values_to_json fail"); + return NULL; + } + + neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, mqtt, + neu_json_encode_mqtt_resp, &json_str); + + if (json.tags) { + free(json.tags); + } + return json_str; +} + +static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, + neu_json_read_req_t *req) +{ + plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group, + req->node); + + neu_reqresp_head_t header = { 0 }; + header.ctx = mqtt; + header.type = NEU_REQ_READ_GROUP; + neu_req_read_group_t cmd = { 0 }; + strcpy(cmd.driver, req->node); + strcpy(cmd.group, req->group); + if (0 != neu_plugin_op(plugin, header, &cmd)) { + plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail"); + return -1; + } + + return 0; +} + int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json, neu_resp_error_t *data) { @@ -185,13 +231,83 @@ int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json, return 0; } +void handle_read_req(const uint8_t *payload, uint32_t len, void *data) +{ + int rv = 0; + neu_plugin_t * plugin = data; + neu_json_read_req_t *req = NULL; + + char *json_str = malloc(len + 1); + if (NULL == json_str) { + return; + } + + memcpy(json_str, payload, len); + json_str[len] = '\0'; + + neu_json_mqtt_t *mqtt = NULL; + rv = neu_json_decode_mqtt_req(json_str, &mqtt); + if (0 != rv) { + plog_error(plugin, "neu_json_decode_mqtt_req failed"); + free(json_str); + return; + } + + rv = neu_json_decode_read_req(json_str, &req); + if (0 != rv) { + plog_error(plugin, "neu_json_decode_read_req fail"); + neu_json_decode_mqtt_req_free(mqtt); + free(json_str); + return; + } + + rv = send_read_req(plugin, mqtt, req); + if (0 != rv) { + neu_json_decode_mqtt_req_free(mqtt); + } + + neu_json_decode_read_req_free(req); + free(json_str); +} + int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json, neu_resp_read_group_t *data) { - (void) plugin; - (void) mqtt_json; - (void) data; - return 0; + int rv = 0; + char *json_str = NULL; + + if (NULL == plugin->client) { + rv = NEU_ERR_MQTT_IS_NULL; + goto end; + } + + if (!neu_mqtt_client_is_connected(plugin->client)) { + rv = NEU_ERR_MQTT_FAILURE; + goto end; + } + + json_str = generate_read_resp_json(plugin, mqtt_json, data); + if (NULL == json_str) { + plog_error(plugin, "generate read resp json fail"); + rv = NEU_ERR_EINTERNAL; + goto end; + } + + const char * topic = plugin->read_resp_topic; + neu_mqtt_qos_e qos = NEU_MQTT_QOS0; + rv = neu_mqtt_client_publish(plugin->client, qos, topic, json_str, + strlen(json_str)); + if (0 != rv) { + plog_error(plugin, "pub [%s, QoS%d] error", topic, qos); + rv = NEU_ERR_MQTT_PUBLISH_FAILURE; + goto end; + } + +end: + neu_json_decode_mqtt_req_free(mqtt_json); + free(json_str); + free(data->tags); + return rv; } int handle_trans_data(neu_plugin_t * plugin, diff --git a/plugins/nano-mqtt/mqtt_handle.h b/plugins/nano-mqtt/mqtt_handle.h index fd32b861..e977046b 100644 --- a/plugins/nano-mqtt/mqtt_handle.h +++ b/plugins/nano-mqtt/mqtt_handle.h @@ -29,6 +29,8 @@ extern "C" { int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json, neu_resp_error_t *data); +void handle_read_req(const uint8_t *payload, uint32_t len, void *data); + int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json, neu_resp_read_group_t *data); diff --git a/plugins/nano-mqtt/mqtt_plugin.c b/plugins/nano-mqtt/mqtt_plugin.c index 1c39b6b1..c895ff37 100644 --- a/plugins/nano-mqtt/mqtt_plugin.c +++ b/plugins/nano-mqtt/mqtt_plugin.c @@ -17,9 +17,11 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#include "mqtt_plugin.h" +#include "utils/asprintf.h" + #include "mqtt_config.h" #include "mqtt_handle.h" +#include "mqtt_plugin.h" const neu_plugin_module_t neu_plugin_module; @@ -69,6 +71,12 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin) neu_mqtt_client_free(plugin->client); plugin->client = NULL; } + + free(plugin->read_req_topic); + plugin->read_req_topic = NULL; + free(plugin->read_resp_topic); + plugin->read_resp_topic = NULL; + plog_info(plugin, "uninitialize plugin `%s` success", neu_plugin_module.module_name); return NEU_ERR_SUCCESS; @@ -185,6 +193,30 @@ error: return rv; } +static int create_topic(neu_plugin_t *plugin) +{ + if (plugin->read_req_topic) { + // topics already created + return 0; + } + + neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req", + plugin->common.name); + if (NULL == plugin->read_req_topic) { + return -1; + } + + neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp", + plugin->common.name); + if (NULL == plugin->read_resp_topic) { + free(plugin->read_req_topic); + plugin->read_req_topic = NULL; + return -1; + } + + return 0; +} + static int mqtt_plugin_start(neu_plugin_t *plugin) { int rv = 0; @@ -202,6 +234,20 @@ static int mqtt_plugin_start(neu_plugin_t *plugin) goto end; } + if (0 != create_topic(plugin)) { + plog_error(plugin, "create topics fail"); + rv = NEU_ERR_EINTERNAL; + goto end; + } + + if (0 != + neu_mqtt_client_subscribe(plugin->client, 0, plugin->read_req_topic, + handle_read_req, plugin)) { + plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic); + rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE; + goto end; + } + end: if (0 == rv) { plog_info(plugin, "start plugin `%s` success", plugin_name); diff --git a/plugins/nano-mqtt/mqtt_plugin.h b/plugins/nano-mqtt/mqtt_plugin.h index d783b60c..01b1bd84 100644 --- a/plugins/nano-mqtt/mqtt_plugin.h +++ b/plugins/nano-mqtt/mqtt_plugin.h @@ -33,6 +33,8 @@ struct neu_plugin { neu_plugin_common_t common; mqtt_config_t config; neu_mqtt_client_t * client; + char * read_req_topic; + char * read_resp_topic; }; #ifdef __cplusplus