diff --git a/include/neuron/json/neu_json_fn.h b/include/neuron/json/neu_json_fn.h index d16b9e14..49a943c8 100644 --- a/include/neuron/json/neu_json_fn.h +++ b/include/neuron/json/neu_json_fn.h @@ -37,9 +37,6 @@ int neu_json_encode_with_mqtt(void *param, neu_json_encode_fn fn, void *mqtt_param, neu_json_encode_fn mqtt_fn, char **result); -int neu_json_encode_setting_with_mqtt(uint32_t node_id, char *setting, - void *mqtt_param, char **result); - #define NEU_JSON_RESPONSE_ERROR(err, func) \ { \ neu_json_error_resp_t error_code = { 0 }; \ diff --git a/include/neuron/json/neu_json_mqtt.h b/include/neuron/json/neu_json_mqtt.h index 6fad94a6..a751ddae 100644 --- a/include/neuron/json/neu_json_mqtt.h +++ b/include/neuron/json/neu_json_mqtt.h @@ -33,7 +33,6 @@ extern "C" { typedef struct { char *uuid; - char *command; } neu_json_mqtt_t; int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result); diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index a623ea3b..3dffbbfa 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -22,11 +22,12 @@ #include "read_write.h" #include "utils/log.h" +#include "../mqtt.h" #include "command.h" // Read tags with group config -static int read_response(mqtt_response_t *response, neu_plugin_t *plugin, - neu_json_mqtt_t *mqtt, char *json_str) +static int read_request(mqtt_response_t *response, neu_plugin_t *plugin, + neu_json_mqtt_t *mqtt, char *json_str) { UNUSED(response); @@ -41,8 +42,8 @@ static int read_response(mqtt_response_t *response, neu_plugin_t *plugin, } // Write tags with group config -static int write_response(mqtt_response_t *response, neu_plugin_t *plugin, - neu_json_mqtt_t *mqtt, char *json_str) +static int write_request(mqtt_response_t *response, neu_plugin_t *plugin, + neu_json_mqtt_t *mqtt, char *json_str) { UNUSED(response); @@ -81,45 +82,50 @@ 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); - zlog_error(neuron, "JSON parsing mqtt failed"); + plog_error(plugin, "json parsing mqtt failed"); + + free(json_str); return; } switch (response->type) { case TOPIC_TYPE_READ: { - rc = read_response(response, plugin, mqtt, json_str); + rc = read_request(response, plugin, mqtt, json_str); + if (0 != rc) { + neu_json_decode_mqtt_req_free(mqtt); + } break; } case TOPIC_TYPE_WRITE: { - rc = write_response(response, plugin, mqtt, json_str); + rc = write_request(response, plugin, mqtt, json_str); + if (0 != rc) { + neu_json_decode_mqtt_req_free(mqtt); + } break; } default: - assert(1 == 0); + plog_error(plugin, "invalid topic type"); break; } - if (NULL != json_str) { - free(json_str); - } - - neu_json_decode_mqtt_req_free(mqtt); + 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 3243c848..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,8 +37,8 @@ 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) { - zlog_info(neuron, "write uuid:%s, group:%s, node:%s", mqtt->uuid, - req->group, req->node); + 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; @@ -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); @@ -154,16 +156,23 @@ char *command_rw_read_once_response(neu_reqresp_head_t * head, uint16_t len = data->n_tag; char * json_str = NULL; neu_json_read_resp_t json = { 0 }; + + 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); clean_read_response_json(&json); + neu_json_decode_mqtt_req_free(head->ctx); 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; @@ -193,7 +202,7 @@ 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) { - zlog_info(neuron, "write uuid:%s, group:%s, node:%s", mqtt->uuid, + plog_info(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid, req->group, req->node); neu_reqresp_head_t header = { 0 }; @@ -236,12 +245,16 @@ 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; + + 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); return json_str; } 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 } diff --git a/src/parser/neu_json_fn.c b/src/parser/neu_json_fn.c index bb4f5a5d..cf2ca8bf 100644 --- a/src/parser/neu_json_fn.c +++ b/src/parser/neu_json_fn.c @@ -67,76 +67,6 @@ int neu_json_encode_with_mqtt(void *param, neu_json_encode_fn fn, return neu_json_encode(object, result); } -int neu_json_encode_setting_with_mqtt(uint32_t node_id, char *setting, - void *mqtt_param, char **result) -{ - neu_json_mqtt_t *mqtt = mqtt_param; - char * params = strstr(setting, "\"params\""); - - if (NULL == params) { - return -1; - } - - int i = sizeof("\"params\""); - int unmatch = 0; - - while (params[i]) { - switch (params[i]) { - case '{': - ++unmatch; // fall through - case ':': - case ' ': - case '\t': - case '\n': - case '\r': - ++i; - break; - default: - return -1; - } - if (unmatch) { - break; - } - } - - while (params[i]) { - if ('{' == params[i]) { - ++unmatch; - } else if ('}' == params[i]) { - --unmatch; - } - ++i; - if (0 == unmatch) { - break; - } - } - - if (unmatch) { - return -1; - } - - // At this point, the string started at `params` with length `i` is the - // the settings that should present in the response. - // Note that this is not an elegant solution due to the limitation of - // the current implementation. - - const char *fmt = "{\"node_id\": %u, \"uuid\": \"%s\", " - "\"command\":\"%s\", " - "%.*s, \"error\": 0}"; - // plus 1 for '\0' - int size = 1 + - snprintf(NULL, 0, fmt, node_id, mqtt->uuid, mqtt->command, i, params); - - *result = calloc(size, sizeof(char)); - if (NULL == *result) { - return -1; - } - - snprintf(*result, size, fmt, node_id, mqtt->uuid, mqtt->command, i, params); - - return 0; -} - int neu_parse_param(char *buf, char **err_param, int n, neu_json_elem_t *ele, ...) { diff --git a/src/parser/neu_json_mqtt.c b/src/parser/neu_json_mqtt.c index 9443302a..0d3d0c14 100644 --- a/src/parser/neu_json_mqtt.c +++ b/src/parser/neu_json_mqtt.c @@ -39,11 +39,6 @@ int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result) .t = NEU_JSON_STR, } }; - neu_json_elem_t elems_command[] = { { - .name = "command", - .t = NEU_JSON_STR, - } }; - ret = neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elems_uuid), elems_uuid); if (ret == 0) { req->uuid = elems_uuid[0].v.val_str; @@ -51,14 +46,6 @@ int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result) goto decode_fail; } - ret = - neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elems_command), elems_command); - if (ret == 0) { - req->command = elems_command[0].v.val_str; - } else { - req->command = strdup(""); - } - *result = req; return 0; @@ -79,10 +66,6 @@ void neu_json_decode_mqtt_req_free(neu_json_mqtt_t *req) free(req->uuid); } - if (NULL != req->command) { - free(req->command); - } - free(req); }