From 41dec9b209ce8b52ef8a07820415f55ff25eead9 Mon Sep 17 00:00:00 2001 From: gc87 Date: Fri, 17 Jun 2022 18:46:30 +0800 Subject: [PATCH] fix(mqtt): plog to mqtt --- plugins/mqtt/command/command.c | 21 +++++----- plugins/mqtt/command/command.h | 8 ++-- plugins/mqtt/command/read_write.c | 25 +++++++----- plugins/mqtt/command/read_write.h | 10 +++-- plugins/mqtt/mqtt.c | 65 +++++++++++++++++-------------- plugins/mqtt/mqtt.h | 21 ++++------ 6 files changed, 83 insertions(+), 67 deletions(-) diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index f9d6249a..3dffbbfa 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -22,6 +22,7 @@ #include "read_write.h" #include "utils/log.h" +#include "../mqtt.h" #include "command.h" // Read tags with group config @@ -81,7 +82,8 @@ void command_response_handle(mqtt_response_t *response) // neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, mqtt, // neu_json_encode_mqtt_resp, &ret_str); - nlog_error("json parsing mqtt failed"); + plog_error(plugin, "json parsing mqtt failed"); + free(json_str); return; } @@ -103,26 +105,27 @@ void command_response_handle(mqtt_response_t *response) break; } default: - nlog_error("invalid topic type"); + plog_error(plugin, "invalid topic type"); break; } free(json_str); } -char *command_read_once_response(neu_reqresp_head_t * head, +char *command_read_once_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, neu_resp_read_group_t *data, int format) { - return command_rw_read_once_response(head, data, format); - return NULL; + return command_rw_read_once_response(plugin, head, data, format); } -char *command_read_periodic_response(neu_reqresp_trans_data_t *data, int format) +char *command_read_periodic_response(neu_plugin_t * plugin, + neu_reqresp_trans_data_t *data, int format) { - return command_rw_read_periodic_response(data, format); + return command_rw_read_periodic_response(plugin, data, format); } -char *command_write_response(neu_reqresp_head_t *head, neu_resp_error_t *data) +char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, + neu_resp_error_t *data) { - return command_rw_write_response(head, data); + return command_rw_write_response(plugin, head, data); } \ No newline at end of file diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index cd8d4f20..dd2cb9c8 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -46,13 +46,15 @@ typedef struct { void command_response_handle(mqtt_response_t *response); -char *command_read_once_response(neu_reqresp_head_t * head, +char *command_read_once_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, neu_resp_read_group_t *data, int format); -char *command_read_periodic_response(neu_reqresp_trans_data_t *data, +char *command_read_periodic_response(neu_plugin_t * plugin, + neu_reqresp_trans_data_t *data, int format); -char *command_write_response(neu_reqresp_head_t *head, neu_resp_error_t *data); +char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, + neu_resp_error_t *data); #ifdef __cplusplus } diff --git a/plugins/mqtt/command/read_write.c b/plugins/mqtt/command/read_write.c index fc0ce120..f024ca6f 100644 --- a/plugins/mqtt/command/read_write.c +++ b/plugins/mqtt/command/read_write.c @@ -22,6 +22,7 @@ #include #include +#include "../mqtt.h" #include "read_write.h" #include "utils/log.h" @@ -36,7 +37,7 @@ static uint64_t current_time() int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_read_req_t *req) { - nlog_info("read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group, + plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group, req->node); neu_reqresp_head_t header = { 0 }; @@ -145,7 +146,8 @@ static void clean_read_response_json(neu_json_read_resp_t *json) free(json->tags); } -char *command_rw_read_once_response(neu_reqresp_head_t * head, +char *command_rw_read_once_response(neu_plugin_t * plugin, + neu_reqresp_head_t * head, neu_resp_read_group_t *data, int format) { UNUSED(format); @@ -155,7 +157,8 @@ char *command_rw_read_once_response(neu_reqresp_head_t * head, char * json_str = NULL; neu_json_read_resp_t json = { 0 }; - nlog_info("read resp uuid: %s", ((neu_json_mqtt_t *) head->ctx)->uuid); + plog_info(plugin, "read resp uuid: %s", + ((neu_json_mqtt_t *) head->ctx)->uuid); wrap_read_response_json(tags, len, &json); neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, head->ctx, neu_json_encode_mqtt_resp, &json_str); @@ -164,9 +167,12 @@ char *command_rw_read_once_response(neu_reqresp_head_t * head, return json_str; } -char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data, +char *command_rw_read_periodic_response(neu_plugin_t * plugin, + neu_reqresp_trans_data_t *data, int format) { + UNUSED(plugin); + neu_resp_tag_value_t * tags = data->tags; uint16_t len = data->n_tag; char * json_str = NULL; @@ -196,8 +202,8 @@ char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data, int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_write_req_t *req) { - nlog_info("write uuid:%s, group:%s, node:%s", mqtt->uuid, req->group, - req->node); + plog_info(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid, + req->group, req->node); neu_reqresp_head_t header = { 0 }; neu_req_write_tag_t cmd = { 0 }; @@ -239,13 +245,14 @@ int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, return 0; } -char *command_rw_write_response(neu_reqresp_head_t *head, - neu_resp_error_t * data) +char *command_rw_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, + neu_resp_error_t *data) { neu_json_error_resp_t error = { .error = data->error }; char * json_str = NULL; - nlog_info("write resp uuid: %s", ((neu_json_mqtt_t *) head->ctx)->uuid); + plog_info(plugin, "write resp uuid: %s", + ((neu_json_mqtt_t *) head->ctx)->uuid); neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, head->ctx, neu_json_encode_mqtt_resp, &json_str); neu_json_decode_mqtt_req_free(head->ctx); diff --git a/plugins/mqtt/command/read_write.h b/plugins/mqtt/command/read_write.h index 70d3e513..1a05c580 100644 --- a/plugins/mqtt/command/read_write.h +++ b/plugins/mqtt/command/read_write.h @@ -31,17 +31,19 @@ extern "C" { int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_read_req_t *req); -char *command_rw_read_once_response(neu_reqresp_head_t * head, +char *command_rw_read_once_response(neu_plugin_t * plugin, + neu_reqresp_head_t * head, neu_resp_read_group_t *data, int format); -char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data, +char *command_rw_read_periodic_response(neu_plugin_t * plugin, + neu_reqresp_trans_data_t *data, int format); int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_write_req_t *req); -char *command_rw_write_response(neu_reqresp_head_t *head, - neu_resp_error_t * data); +char *command_rw_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, + neu_resp_error_t *data); #ifdef __cplusplus } diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 516cf45b..ac99767c 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -18,15 +18,25 @@ **/ #include -#include -#include -#include #include "command/command.h" #include "mqtt.h" -#include "neuron.h" #include "util.h" +#define INTERVAL 100000U +#define TIMEOUT 3000U + +#define TOPIC_READ_REQ "neuron/%s/read/req" +#define TOPIC_WRITE_REQ "neuron/%s/write/req" + +#define TOPIC_READ_RES "neuron/%s/read/resp" +#define TOPIC_WRITE_RES "neuron/%s/write/resp" +#define TOPIC_UPLOAD_RES "neuron/%s/upload" + +#define QOS0 0 +#define QOS1 1 +#define QOS2 2 + const neu_plugin_module_t neu_plugin_module; struct topic_pair { @@ -37,20 +47,15 @@ struct topic_pair { int type; }; -typedef struct mqtt_routine { +struct mqtt_routine { neu_plugin_t * plugin; neu_mqtt_option_t option; neu_mqtt_client_t client; UT_array * topics; -} mqtt_routine_t; - -struct neu_plugin { - neu_plugin_common_t common; - bool running; - mqtt_routine_t * routine; - char * config; }; +static neu_plugin_t *plugin_log = NULL; + static char *topics_format(char *format, char *name) { char temp[256] = { '\0' }; @@ -73,8 +78,8 @@ static void topics_add(UT_array *topics, char *request, int qos_request, pair->qos_response = qos_response; pair->type = type; utarray_push_back(topics, &pair); - nlog_info("add topic-req:%s, topic-res:%s, count:%d", pair->topic_request, - pair->topic_response, utarray_len(topics)); + plog_info(plugin_log, "add topic-req:%s, topic-res:%s, count:%d", + pair->topic_request, pair->topic_response, utarray_len(topics)); } static void topics_cleanup(UT_array *topics) @@ -84,8 +89,8 @@ static void topics_cleanup(UT_array *topics) NULL != pair; pair = (struct topic_pair **) utarray_next(topics, pair)) { - nlog_info("remove topic-req:%s, topic-res:%s", (*pair)->topic_request, - (*pair)->topic_response); + plog_info(plugin_log, "remove topic-req:%s, topic-res:%s", + (*pair)->topic_request, (*pair)->topic_response); if (NULL != (*pair)->topic_request) { free((*pair)->topic_request); } @@ -257,6 +262,7 @@ static neu_plugin_t *mqtt_plugin_open(void) { neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t)); neu_plugin_common_init(&plugin->common); + plugin_log = plugin; // for plog return plugin; } @@ -267,6 +273,7 @@ static int mqtt_plugin_close(neu_plugin_t *plugin) const char *name = neu_plugin_module.module_name; plog_info(plugin, "success to free plugin:%s", name); + plugin_log = NULL; free(plugin); return NEU_ERR_SUCCESS; } @@ -382,13 +389,13 @@ static neu_err_code_e write_response(neu_plugin_t * plugin, return NEU_ERR_MQTT_FAILURE; } - mqtt_routine_t * routine = plugin->routine; - int type = TOPIC_TYPE_READ; - struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json_str = command_write_response(head, write_data); - const char * topic = pair->topic_response; - const int qos = pair->qos_response; - neu_err_code_e error = + mqtt_routine_t * routine = plugin->routine; + int type = TOPIC_TYPE_READ; + struct topic_pair *pair = topics_find_type(routine->topics, type); + char * json_str = command_write_response(plugin, head, write_data); + const char * topic = pair->topic_response; + const int qos = pair->qos_response; + neu_err_code_e error = neu_mqtt_client_publish(routine->client, topic, qos, (unsigned char *) json_str, strlen(json_str)); if (NEU_ERR_SUCCESS != error) { @@ -414,10 +421,10 @@ static neu_err_code_e read_response(neu_plugin_t * plugin, mqtt_routine_t * routine = plugin->routine; int type = TOPIC_TYPE_READ; struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json_str = - command_read_once_response(head, read_data, routine->option.format); - const char * topic = pair->topic_response; - const int qos = pair->qos_response; + char * json_str = command_read_once_response(plugin, head, read_data, + routine->option.format); + const char *topic = pair->topic_response; + const int qos = pair->qos_response; neu_err_code_e error = neu_mqtt_client_publish(routine->client, topic, qos, (unsigned char *) json_str, strlen(json_str)); @@ -443,8 +450,8 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) mqtt_routine_t * routine = plugin->routine; int type = TOPIC_TYPE_UPLOAD; struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json_str = - command_read_periodic_response(trans_data, routine->option.format); + char *json_str = command_read_periodic_response(plugin, trans_data, + routine->option.format); if (NULL == json_str) { return NEU_ERR_MQTT_FAILURE; } diff --git a/plugins/mqtt/mqtt.h b/plugins/mqtt/mqtt.h index 15383f65..6bac6bb5 100644 --- a/plugins/mqtt/mqtt.h +++ b/plugins/mqtt/mqtt.h @@ -25,21 +25,16 @@ extern "C" { #endif #include +#include -#define UNUSED(x) (void) (x) -#define INTERVAL 100000U -#define TIMEOUT 3000U +typedef struct mqtt_routine mqtt_routine_t; -#define TOPIC_READ_REQ "neuron/%s/read/req" -#define TOPIC_WRITE_REQ "neuron/%s/write/req" - -#define TOPIC_READ_RES "neuron/%s/read/resp" -#define TOPIC_WRITE_RES "neuron/%s/write/resp" -#define TOPIC_UPLOAD_RES "neuron/%s/upload" - -#define QOS0 0 -#define QOS1 1 -#define QOS2 2 +struct neu_plugin { + neu_plugin_common_t common; + bool running; + mqtt_routine_t * routine; + char * config; +}; #ifdef __cplusplus }