From efd8083a7564f85ea543643e49cfed9e411258af Mon Sep 17 00:00:00 2001 From: gc87 Date: Wed, 6 Jul 2022 17:10:36 +0800 Subject: [PATCH 1/9] fix(mqtt): bio should retry flag clear --- src/connection/mqtt_c_client.c | 1 + 1 file changed, 1 insertion(+) diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index 4e478f2a..9955403a 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -274,6 +274,7 @@ static BIO *bio_tcp_create(const char *addr, const char *port) return NULL; } + BIO_clear_flags(bio, BIO_FLAGS_SHOULD_RETRY); BIO_set_nbio(bio, 1); BIO_set_conn_port(bio, port); return bio; From 649de85b6179257619c2ee7a7e3a74373e5873d8 Mon Sep 17 00:00:00 2001 From: gc87 Date: Tue, 12 Jul 2022 14:16:41 +0800 Subject: [PATCH 2/9] feat(mqtt): add heartbeat setting --- include/neuron/connection/mqtt_client_intf.h | 1 + plugins/mqtt/command/command.h | 1 + plugins/mqtt/mqtt.c | 19 +++++++++-- plugins/mqtt/mqtt.json | 10 ++++++ plugins/mqtt/util.c | 36 ++++++++++++-------- tests/mqtt_test.cc | 3 +- 6 files changed, 53 insertions(+), 17 deletions(-) diff --git a/include/neuron/connection/mqtt_client_intf.h b/include/neuron/connection/mqtt_client_intf.h index 6801c140..4d8fad73 100644 --- a/include/neuron/connection/mqtt_client_intf.h +++ b/include/neuron/connection/mqtt_client_intf.h @@ -60,6 +60,7 @@ typedef struct { /*application*/ int format; char * upload_topic; + char * heartbeat_topic; char * group_id; state_update state_update_func; zlog_category_t *log; diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index dd2cb9c8..fce520a0 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -32,6 +32,7 @@ enum topic_type { TOPIC_TYPE_READ, TOPIC_TYPE_WRITE, TOPIC_TYPE_UPLOAD, + TOPIC_TYPE_HEARTBEAT, }; typedef struct { diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 31d64344..3c11f954 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -32,6 +32,7 @@ #define TOPIC_READ_RES "neuron/%s/read/resp" #define TOPIC_WRITE_RES "neuron/%s/write/resp" #define TOPIC_UPLOAD_RES "neuron/%s/upload" +#define TOPIC_HEARTBEAT_RES "neuron/%s/heartbeat" #define QOS0 0 #define QOS1 1 @@ -103,7 +104,8 @@ static void topics_cleanup(UT_array *topics) } } -static void topics_generate(UT_array *topics, char *name, char *upload_topic) +static void topics_generate(UT_array *topics, char *name, char *upload_topic, + char *heartbeat_topic) { char *read_req = topics_format(TOPIC_READ_REQ, name); char *read_res = topics_format(TOPIC_READ_RES, name); @@ -123,6 +125,18 @@ static void topics_generate(UT_array *topics, char *name, char *upload_topic) } topics_add(topics, upload_req, QOS0, upload_res, QOS0, TOPIC_TYPE_UPLOAD); + + // HEARTBEAT TOPIC SETTING + char *heartbeat_req = NULL; + char *heartbeat_res = NULL; + if (NULL != heartbeat_topic && 0 < strlen(heartbeat_topic)) { + heartbeat_res = strdup(heartbeat_topic); + } else { + heartbeat_res = topics_format(TOPIC_HEARTBEAT_RES, name); + } + + topics_add(topics, heartbeat_req, QOS0, heartbeat_res, QOS0, + TOPIC_TYPE_HEARTBEAT); } static struct topic_pair *topics_find_topic(UT_array *topics, const char *topic) @@ -237,7 +251,8 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, UT_icd ptr_icd = { sizeof(struct topic_pair *), NULL, NULL, NULL }; utarray_new(routine->topics, &ptr_icd); topics_generate(routine->topics, routine->option.clientid, - routine->option.upload_topic); + routine->option.upload_topic, + routine->option.heartbeat_topic); topics_subscribe(routine->topics, routine->client); return routine; } diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index 869d8d16..a454ede0 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -19,6 +19,16 @@ "length": 255 } }, + "heartbeat-topic": { + "name": "heartbeat topic", + "description": "User-defined heartbeat topic", + "type": "string", + "attribute": "optional", + "default": "", + "valid": { + "length": 255 + } + }, "format": { "name": "upload format", "description": "Upload format on mqtt", diff --git a/plugins/mqtt/util.c b/plugins/mqtt/util.c index b66ab0a5..753fd250 100644 --- a/plugins/mqtt/util.c +++ b/plugins/mqtt/util.c @@ -40,6 +40,11 @@ void mqtt_option_uninit(neu_mqtt_option_t *option) option->upload_topic = NULL; } + if (NULL != option->heartbeat_topic) { + free(option->heartbeat_topic); + option->heartbeat_topic = NULL; + } + if (NULL != option->connection) { free(option->connection); option->connection = NULL; @@ -92,20 +97,22 @@ int mqtt_option_init(char *config, neu_mqtt_option_t *option) return -1; } - int ret = 0; - char * error = NULL; - neu_json_elem_t id = { .name = "client-id", .t = NEU_JSON_STR }; - neu_json_elem_t upload = { .name = "upload-topic", .t = NEU_JSON_STR }; - neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT }; - neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL }; - neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; - neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT }; - neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR }; - neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR }; - neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR }; - neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR }; - neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR }; - neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR }; + int ret = 0; + char * error = NULL; + neu_json_elem_t id = { .name = "client-id", .t = NEU_JSON_STR }; + neu_json_elem_t upload = { .name = "upload-topic", .t = NEU_JSON_STR }; + neu_json_elem_t heartbeat = { .name = "heartbeat-topic", + .t = NEU_JSON_STR }; + neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT }; + neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL }; + neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; + neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT }; + neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR }; + neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR }; + neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR }; + neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR }; + neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR }; + neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR }; ret = neu_parse_param(config, &error, 12, &id, &upload, &format, &ssl, &host, &port, &username, &password, &ca, &cert, &key, @@ -135,6 +142,7 @@ int mqtt_option_init(char *config, neu_mqtt_option_t *option) snprintf(option->port, 10, "%d", p); option->clientid = id.v.val_str; option->upload_topic = upload.v.val_str; + option->heartbeat_topic = heartbeat.v.val_str; option->format = format.v.val_int; option->username = username.v.val_str; option->password = password.v.val_str; diff --git a/tests/mqtt_test.cc b/tests/mqtt_test.cc index cc242ae7..6e7a7ef8 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -13,7 +13,8 @@ TEST(MQTTTest, mqtt_option_init_by_config) memset(&option, 0, sizeof(neu_mqtt_option_t)); char *config = (char *) "{\"node_id\":5,\"params\":{\"client-id\":" - "\"upload123\", \"upload-topic\":\"\", \"format\": " + "\"upload123\", \"upload-topic\":\"\", " + "\"heartbeat-topic\":\"\", \"format\": " "0,\"ssl\":false,\"host\":\"192.168.10." "116\",\"port\":1883,\"username\":\"\",\"password\":" "\"\",\"ca\":\"\", \"cert\":\"\", \"key\":\"\", " From 1c9d6a4b68458b0d67c52f109ad5a72e0c528c8f Mon Sep 17 00:00:00 2001 From: gc87 Date: Tue, 12 Jul 2022 15:26:02 +0800 Subject: [PATCH 3/9] feat(mqtt): add heartbeat timer --- plugins/mqtt/mqtt.c | 55 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 53 insertions(+), 2 deletions(-) diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 3c11f954..1cf722bf 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -18,6 +18,8 @@ **/ #include +#include +#include #include "command/command.h" #include "mqtt.h" @@ -53,6 +55,7 @@ struct mqtt_routine { neu_mqtt_option_t option; neu_mqtt_client_t client; UT_array * topics; + nng_aio * aio; }; static neu_plugin_t *plugin_log = NULL; @@ -212,6 +215,44 @@ static void mqtt_routine_state(void *context, int state) } } +static void mqtt_routine_heartbeat(nng_aio *aio, void *arg, int code) +{ + mqtt_routine_t *routine = (mqtt_routine_t *) arg; + + switch (code) { + case NNG_ETIMEDOUT: { + int type = TOPIC_TYPE_HEARTBEAT; + struct topic_pair *pair = topics_find_type(routine->topics, type); + char * json_str = "test"; + 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) { + plog_error(plugin_log, + "heartbeat publish error code :%d, topoic:%s", error, + topic); + // free(json_str); + } + + plog_info(plugin_log, "topic:%s, qos:%d, payload:%s", topic, qos, + json_str); + // free(json_str); + + nng_aio_set_timeout(aio, 1000); + nng_aio_defer(aio, mqtt_routine_heartbeat, routine); + break; + } + case NNG_ECANCELED: { + plog_info(plugin_log, "aio:%p, cancel", aio); + break; + } + default: + plog_warn(plugin_log, "aio:%p, skip error:%d", aio, code); + } +} + static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, const char * config) { @@ -233,9 +274,9 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, routine->option.state_update_func = mqtt_routine_state; routine->option.log = plugin->common.log; - neu_mqtt_client_t *client = (neu_mqtt_client_t *) &routine->client; + neu_mqtt_client_t *client_p = (neu_mqtt_client_t *) &routine->client; neu_err_code_e error = NEU_ERR_SUCCESS; - error = neu_mqtt_client_open(client, &routine->option, plugin); + error = neu_mqtt_client_open(client_p, &routine->option, plugin); if (NEU_ERR_SUCCESS != error) { mqtt_option_uninit(&routine->option); free(routine); @@ -254,6 +295,12 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, routine->option.upload_topic, routine->option.heartbeat_topic); topics_subscribe(routine->topics, routine->client); + + nng_aio_alloc(&routine->aio, NULL, routine); + nng_aio_wait(routine->aio); + nng_aio_set_timeout(routine->aio, 1000); + nng_aio_defer(routine->aio, mqtt_routine_heartbeat, routine); + return routine; } @@ -261,6 +308,10 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) { assert(NULL != routine); + // stop heartbeat sent + nng_aio_cancel(routine->aio); + nng_aio_free(routine->aio); + // stop recevied neu_mqtt_client_suspend(routine->client); From 1f9e3ea0db1e2356527d0cd93ffa1eeda19ca7d7 Mon Sep 17 00:00:00 2001 From: gc87 Date: Tue, 12 Jul 2022 22:15:52 +0800 Subject: [PATCH 4/9] feat(mqtt): add node state cache & add hb command --- plugins/mqtt/CMakeLists.txt | 1 + plugins/mqtt/command/command.c | 6 ++ plugins/mqtt/command/command.h | 2 + plugins/mqtt/command/heartbeat.c | 38 +++++++++++ plugins/mqtt/command/heartbeat.h | 36 +++++++++++ plugins/mqtt/mqtt.c | 108 +++++++++++++++++++++++++++++-- 6 files changed, 187 insertions(+), 4 deletions(-) create mode 100644 plugins/mqtt/command/heartbeat.c create mode 100644 plugins/mqtt/command/heartbeat.h diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index 8549d8c1..983c0145 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -10,6 +10,7 @@ set(src mqtt.c util.c command/read_write.c + command/heartbeat.c command/command.c) add_library(${PROJECT_NAME} SHARED ${src}) diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index 3dffbbfa..d29d2d12 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -19,6 +19,7 @@ #include #include +#include "heartbeat.h" #include "read_write.h" #include "utils/log.h" @@ -128,4 +129,9 @@ char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, neu_resp_error_t *data) { return command_rw_write_response(plugin, head, data); +} + +char *command_heartbeat_response(neu_plugin_t *plugin) +{ + return command_heartbeat_generate(plugin); } \ No newline at end of file diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index fce520a0..7035b917 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -57,6 +57,8 @@ char *command_read_periodic_response(neu_plugin_t * plugin, char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, neu_resp_error_t *data); +char *command_heartbeat_response(neu_plugin_t *plugin); + #ifdef __cplusplus } #endif diff --git a/plugins/mqtt/command/heartbeat.c b/plugins/mqtt/command/heartbeat.c new file mode 100644 index 00000000..01a34a51 --- /dev/null +++ b/plugins/mqtt/command/heartbeat.c @@ -0,0 +1,38 @@ +/** + * 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 "../mqtt.h" +#include "heartbeat.h" +#include "utils/log.h" +#include "version.h" + +char *command_heartbeat_generate(neu_plugin_t *plugin) +{ + UNUSED(plugin); + + // const char *version = NEURON_VERSION; + // const char *revision = NEURON_GIT_REV NEURON_GIT_DIFF; + // const char * build_date = NEURON_BUILD_DATE; + + return NULL; +} \ No newline at end of file diff --git a/plugins/mqtt/command/heartbeat.h b/plugins/mqtt/command/heartbeat.h new file mode 100644 index 00000000..4f183939 --- /dev/null +++ b/plugins/mqtt/command/heartbeat.h @@ -0,0 +1,36 @@ +/** + * 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 NEURON_PLUGIN_MQTT_COMMAND_HEARTBEAT +#define NEURON_PLUGIN_MQTT_COMMAND_HEARTBEAT + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "common.h" + +char *command_heartbeat_generate(neu_plugin_t *plugin); + +#ifdef __cplusplus +} +#endif +#endif diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 1cf722bf..5f26a918 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -50,12 +50,20 @@ struct topic_pair { int type; }; +struct node_state { + char node[NEU_NODE_NAME_LEN]; + char plugin[NEU_PLUGIN_NAME_LEN]; + int running; + int link; +}; + struct mqtt_routine { neu_plugin_t * plugin; neu_mqtt_option_t option; neu_mqtt_client_t client; UT_array * topics; nng_aio * aio; + UT_array * states; }; static neu_plugin_t *plugin_log = NULL; @@ -221,6 +229,21 @@ static void mqtt_routine_heartbeat(nng_aio *aio, void *arg, int code) switch (code) { case NNG_ETIMEDOUT: { + // command all node list + neu_req_get_node_t cmd = { 0 }; + neu_reqresp_head_t header = { + .ctx = NULL, + .type = NEU_REQ_GET_NODE, + }; + + cmd.type = NEU_NA_TYPE_APP; + int ret = neu_plugin_op(routine->plugin, header, &cmd); + if (ret != 0) { } + cmd.type = NEU_NA_TYPE_DRIVER; + ret = neu_plugin_op(routine->plugin, header, &cmd); + if (ret != 0) { } + + // publish state data int type = TOPIC_TYPE_HEARTBEAT; struct topic_pair *pair = topics_find_type(routine->topics, type); char * json_str = "test"; @@ -288,14 +311,18 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, plog_info(plugin, "open mqtt client: %s:%s, code:%d", routine->option.host, routine->option.port, error); - routine->plugin = plugin; - UT_icd ptr_icd = { sizeof(struct topic_pair *), NULL, NULL, NULL }; - utarray_new(routine->topics, &ptr_icd); + routine->plugin = plugin; + UT_icd topic_ptr_icd = { sizeof(struct topic_pair *), NULL, NULL, NULL }; + utarray_new(routine->topics, &topic_ptr_icd); topics_generate(routine->topics, routine->option.clientid, routine->option.upload_topic, routine->option.heartbeat_topic); topics_subscribe(routine->topics, routine->client); + UT_icd state_ptr_icd = { sizeof(struct node_state *), NULL, NULL, NULL }; + utarray_new(routine->states, &state_ptr_icd); + + // start heartbeat send nng_aio_alloc(&routine->aio, NULL, routine); nng_aio_wait(routine->aio); nng_aio_set_timeout(routine->aio, 1000); @@ -308,7 +335,9 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) { assert(NULL != routine); - // stop heartbeat sent + utarray_free(routine->states); + + // stop heartbeat send nng_aio_cancel(routine->aio); nng_aio_free(routine->aio); @@ -576,6 +605,69 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) // return NEU_ERR_SUCCESS; // } +static neu_err_code_e node_list_get(neu_plugin_t * plugin, + neu_reqresp_head_t *head, void *data) +{ + UNUSED(head); + + neu_resp_get_node_t *nodes = (neu_resp_get_node_t *) data; + utarray_foreach(nodes->nodes, neu_resp_node_info_t *, info) + { + plog_info(plugin, "%s, %s", info->node, info->plugin); + neu_reqresp_head_t header = { 0 }; + neu_req_get_node_state_t cmd = { 0 }; + header.ctx = strdup(info->node); + header.type = NEU_REQ_GET_NODE_STATE; + + strcpy(cmd.node, info->node); + int ret = neu_plugin_op(plugin, header, &cmd); + if (ret != 0) { } + } + + utarray_free(nodes->nodes); + + return NEU_ERR_SUCCESS; +} + +static neu_err_code_e node_state_update(neu_plugin_t * plugin, + neu_reqresp_head_t *head, void *data) +{ + mqtt_routine_t * routine = plugin->routine; + neu_resp_get_node_state_t *state = (neu_resp_get_node_state_t *) data; + char * node = (char *) head->ctx; + int link = state->state.link; + int running = state->state.running; + plog_debug(plugin, "node:%s, link:%d, running:%d", node, link, running); + + struct node_state *ns = NULL; + for (struct node_state **sp = + (struct node_state **) utarray_front(routine->states); + sp != NULL; + sp = (struct node_state **) utarray_next(routine->states, sp)) { + + // plog_info(plugin, "node:%s, state->link:%d, state->running:%d", + // (char *) head->ctx, state->link, state->running); + + if (0 == strcmp((char *) head->ctx, (*sp)->node)) { + ns = (*sp); + break; + } + } + + if (NULL == ns) { + ns = calloc(1, sizeof(struct node_state)); + strcpy(ns->node, head->ctx); + utarray_push_back(routine->states, &ns); + } + + ns->link = state->state.link; + ns->running = state->state.running; + + free(head->ctx); + + return NEU_ERR_SUCCESS; +} + static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, void *data) { @@ -606,6 +698,14 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, utarray_free(sub->tags); break; } + case NEU_RESP_GET_NODE: { + error = node_list_get(plugin, head, data); + break; + } + case NEU_RESP_GET_NODE_STATE: { + error = node_state_update(plugin, head, data); + break; + } default: error = NEU_ERR_MQTT_FAILURE; break; From 596cd699fd3992221bb9b4b39ea52eb6ed369200 Mon Sep 17 00:00:00 2001 From: gc87 Date: Wed, 13 Jul 2022 16:00:04 +0800 Subject: [PATCH 5/9] feat(mqtt): state json string send --- ft/resource/common.resource | 2 +- include/neuron/json/neu_json_mqtt.h | 19 +++++++++ plugins/mqtt/command/command.c | 4 +- plugins/mqtt/command/command.h | 2 +- plugins/mqtt/command/heartbeat.c | 30 +++++++++++--- plugins/mqtt/command/heartbeat.h | 2 +- plugins/mqtt/mqtt.c | 63 ++++++++++++++++------------- plugins/mqtt/mqtt.h | 7 ++++ plugins/mqtt/util.c | 6 +-- src/parser/neu_json_mqtt.c | 62 ++++++++++++++++++++++++++++ 10 files changed, 154 insertions(+), 43 deletions(-) diff --git a/ft/resource/common.resource b/ft/resource/common.resource index c3032bae..1ffb5467 100644 --- a/ft/resource/common.resource +++ b/ft/resource/common.resource @@ -35,7 +35,7 @@ ${TOPIC_UPLOAD} neuron/${MQTT_CLIENTID}/upload ${MQTT_UUID} 123456-789abc-123456-aabbcc ${MQTT_CONFIG_HOST} 127.0.0.1 ${MQTT_CONFIG_PORT} ${1883} -${MQTT_CONFIG} {"client-id": "${MQTT_CLIENTID}", "upload-topic": "", "format": 1, +${MQTT_CONFIG} {"client-id": "${MQTT_CLIENTID}", "upload-topic": "", "heartbeat-topic": "", "format": 1, ... "ssl": false, "host": "${MQTT_CONFIG_HOST}", "port": ${MQTT_CONFIG_PORT}, ... "username": "", "password": "", "ca":"", "cert":"", "key":"", "keypass":""} diff --git a/include/neuron/json/neu_json_mqtt.h b/include/neuron/json/neu_json_mqtt.h index a751ddae..c091b66d 100644 --- a/include/neuron/json/neu_json_mqtt.h +++ b/include/neuron/json/neu_json_mqtt.h @@ -35,11 +35,30 @@ typedef struct { char *uuid; } neu_json_mqtt_t; +typedef struct { + char *node; + int link; + int running; +} neu_json_node_state_t; + +typedef struct { + char * version; + uint32_t timpstamp; +} neu_json_states_head_t; + +typedef struct { + int n_state; + neu_json_node_state_t *states; +} neu_json_states_t; + int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result); void neu_json_decode_mqtt_req_free(neu_json_mqtt_t *req); int neu_json_encode_mqtt_resp(void *json_object, void *param); +int neu_json_encode_state_header_resp(void *json_object, void *param); +int neu_json_encode_states_resp(void *json_object, void *param); + #ifdef __cplusplus } #endif diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index d29d2d12..c39f6bf8 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -131,7 +131,7 @@ char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, return command_rw_write_response(plugin, head, data); } -char *command_heartbeat_response(neu_plugin_t *plugin) +char *command_heartbeat_response(neu_plugin_t *plugin, UT_array *states) { - return command_heartbeat_generate(plugin); + return command_heartbeat_generate(plugin, states); } \ No newline at end of file diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index 7035b917..be19d464 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -57,7 +57,7 @@ char *command_read_periodic_response(neu_plugin_t * plugin, char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head, neu_resp_error_t *data); -char *command_heartbeat_response(neu_plugin_t *plugin); +char *command_heartbeat_response(neu_plugin_t *plugin, UT_array *states); #ifdef __cplusplus } diff --git a/plugins/mqtt/command/heartbeat.c b/plugins/mqtt/command/heartbeat.c index 01a34a51..e60625fd 100644 --- a/plugins/mqtt/command/heartbeat.c +++ b/plugins/mqtt/command/heartbeat.c @@ -26,13 +26,31 @@ #include "utils/log.h" #include "version.h" -char *command_heartbeat_generate(neu_plugin_t *plugin) +char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states) { - UNUSED(plugin); + char * version = NEURON_VERSION; + neu_json_states_head_t header = { .version = version, + .timpstamp = plugin->common.timestamp }; + neu_json_states_t json = { 0 }; + char * json_str = NULL; + json.n_state = utarray_len(states); + if (0 < json.n_state) { + json.states = calloc(json.n_state, sizeof(neu_json_node_state_t)); + } - // const char *version = NEURON_VERSION; - // const char *revision = NEURON_GIT_REV NEURON_GIT_DIFF; - // const char * build_date = NEURON_BUILD_DATE; + for (int i = 0; i < json.n_state; i++) { + struct node_state **nsp = utarray_eltptr(states, (unsigned int) i); + struct node_state * ns = (*nsp); + json.states[i].node = ns->node; + json.states[i].link = ns->link; + json.states[i].running = ns->running; + } - return NULL; + neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header, + neu_json_encode_state_header_resp, &json_str); + if (0 < json.n_state) { + free(json.states); + } + + return json_str; } \ No newline at end of file diff --git a/plugins/mqtt/command/heartbeat.h b/plugins/mqtt/command/heartbeat.h index 4f183939..5009a0c0 100644 --- a/plugins/mqtt/command/heartbeat.h +++ b/plugins/mqtt/command/heartbeat.h @@ -28,7 +28,7 @@ extern "C" { #include "common.h" -char *command_heartbeat_generate(neu_plugin_t *plugin); +char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states); #ifdef __cplusplus } diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 5f26a918..98faa837 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -50,19 +50,13 @@ struct topic_pair { int type; }; -struct node_state { - char node[NEU_NODE_NAME_LEN]; - char plugin[NEU_PLUGIN_NAME_LEN]; - int running; - int link; -}; - struct mqtt_routine { neu_plugin_t * plugin; neu_mqtt_option_t option; neu_mqtt_client_t client; UT_array * topics; nng_aio * aio; + nng_mtx * mutex; UT_array * states; }; @@ -141,6 +135,7 @@ static void topics_generate(UT_array *topics, char *name, char *upload_topic, char *heartbeat_req = NULL; char *heartbeat_res = NULL; if (NULL != heartbeat_topic && 0 < strlen(heartbeat_topic)) { + nlog_debug("------------->%s", heartbeat_topic); heartbeat_res = strdup(heartbeat_topic); } else { heartbeat_res = topics_format(TOPIC_HEARTBEAT_RES, name); @@ -243,28 +238,33 @@ static void mqtt_routine_heartbeat(nng_aio *aio, void *arg, int code) ret = neu_plugin_op(routine->plugin, header, &cmd); if (ret != 0) { } + nng_aio_set_timeout(aio, 3000); + nng_aio_defer(aio, mqtt_routine_heartbeat, routine); + // publish state data - int type = TOPIC_TYPE_HEARTBEAT; - struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json_str = "test"; - const char * topic = pair->topic_response; - const int qos = pair->qos_response; - neu_err_code_e error = neu_mqtt_client_publish( + nng_mtx_lock(routine->mutex); + int type = TOPIC_TYPE_HEARTBEAT; + struct topic_pair *pair = topics_find_type(routine->topics, type); + char * json_str = + command_heartbeat_response(routine->plugin, routine->states); + if (NULL == json_str) { + nng_mtx_unlock(routine->mutex); + break; + } + + 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) { plog_error(plugin_log, "heartbeat publish error code :%d, topoic:%s", error, topic); - // free(json_str); } - plog_info(plugin_log, "topic:%s, qos:%d, payload:%s", topic, qos, - json_str); - // free(json_str); - - nng_aio_set_timeout(aio, 1000); - nng_aio_defer(aio, mqtt_routine_heartbeat, routine); + free(json_str); + nng_mtx_unlock(routine->mutex); break; } case NNG_ECANCELED: { @@ -323,9 +323,10 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, utarray_new(routine->states, &state_ptr_icd); // start heartbeat send + nng_mtx_alloc(&routine->mutex); nng_aio_alloc(&routine->aio, NULL, routine); nng_aio_wait(routine->aio); - nng_aio_set_timeout(routine->aio, 1000); + nng_aio_set_timeout(routine->aio, 3000); nng_aio_defer(routine->aio, mqtt_routine_heartbeat, routine); return routine; @@ -335,9 +336,18 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) { assert(NULL != routine); + // cleanup node states + for (struct node_state **sp = + (struct node_state **) utarray_front(routine->states); + sp != NULL; + sp = (struct node_state **) utarray_next(routine->states, sp)) { + free((*sp)); + } + utarray_free(routine->states); // stop heartbeat send + nng_mtx_free(routine->mutex); nng_aio_cancel(routine->aio); nng_aio_free(routine->aio); @@ -613,7 +623,7 @@ static neu_err_code_e node_list_get(neu_plugin_t * plugin, neu_resp_get_node_t *nodes = (neu_resp_get_node_t *) data; utarray_foreach(nodes->nodes, neu_resp_node_info_t *, info) { - plog_info(plugin, "%s, %s", info->node, info->plugin); + // plog_info(plugin, "%s, %s", info->node, info->plugin); neu_reqresp_head_t header = { 0 }; neu_req_get_node_state_t cmd = { 0 }; header.ctx = strdup(info->node); @@ -634,20 +644,14 @@ static neu_err_code_e node_state_update(neu_plugin_t * plugin, { mqtt_routine_t * routine = plugin->routine; neu_resp_get_node_state_t *state = (neu_resp_get_node_state_t *) data; - char * node = (char *) head->ctx; - int link = state->state.link; - int running = state->state.running; - plog_debug(plugin, "node:%s, link:%d, running:%d", node, link, running); + nng_mtx_lock(routine->mutex); struct node_state *ns = NULL; for (struct node_state **sp = (struct node_state **) utarray_front(routine->states); sp != NULL; sp = (struct node_state **) utarray_next(routine->states, sp)) { - // plog_info(plugin, "node:%s, state->link:%d, state->running:%d", - // (char *) head->ctx, state->link, state->running); - if (0 == strcmp((char *) head->ctx, (*sp)->node)) { ns = (*sp); break; @@ -662,6 +666,7 @@ static neu_err_code_e node_state_update(neu_plugin_t * plugin, ns->link = state->state.link; ns->running = state->state.running; + nng_mtx_unlock(routine->mutex); free(head->ctx); diff --git a/plugins/mqtt/mqtt.h b/plugins/mqtt/mqtt.h index 6bac6bb5..93818d30 100644 --- a/plugins/mqtt/mqtt.h +++ b/plugins/mqtt/mqtt.h @@ -29,6 +29,13 @@ extern "C" { typedef struct mqtt_routine mqtt_routine_t; +struct node_state { + char node[NEU_NODE_NAME_LEN]; + char plugin[NEU_PLUGIN_NAME_LEN]; + int running; + int link; +}; + struct neu_plugin { neu_plugin_common_t common; bool running; diff --git a/plugins/mqtt/util.c b/plugins/mqtt/util.c index 753fd250..ca73d2e6 100644 --- a/plugins/mqtt/util.c +++ b/plugins/mqtt/util.c @@ -114,9 +114,9 @@ int mqtt_option_init(char *config, neu_mqtt_option_t *option) neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR }; neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR }; - ret = neu_parse_param(config, &error, 12, &id, &upload, &format, &ssl, - &host, &port, &username, &password, &ca, &cert, &key, - &keypass); + ret = neu_parse_param(config, &error, 13, &id, &upload, &heartbeat, &format, + &ssl, &host, &port, &username, &password, &ca, &cert, + &key, &keypass); if (0 != ret) { return ret; } diff --git a/src/parser/neu_json_mqtt.c b/src/parser/neu_json_mqtt.c index 0d3d0c14..8f560997 100644 --- a/src/parser/neu_json_mqtt.c +++ b/src/parser/neu_json_mqtt.c @@ -22,6 +22,7 @@ * It was automatically generated by `json-autotype`. */ +#include #include #include @@ -85,3 +86,64 @@ int neu_json_encode_mqtt_resp(void *json_object, void *param) return ret; } + +int neu_json_encode_state_header_resp(void *json_object, void *param) +{ + int ret = 0; + neu_json_states_head_t *resp = (neu_json_states_head_t *) param; + + neu_json_elem_t resp_elems[] = { { + .name = "version", + .t = NEU_JSON_STR, + .v.val_str = resp->version, + }, + { + .name = "timestamp", + .t = NEU_JSON_INT, + .v.val_int = resp->timpstamp, + } }; + ret = neu_json_encode_field(json_object, resp_elems, + NEU_JSON_ELEM_SIZE(resp_elems)); + + return ret; +} + +int neu_json_encode_states_resp(void *json_object, void *param) +{ + int ret = 0; + neu_json_states_t *resp = (neu_json_states_t *) param; + + void * state_array = neu_json_array(); + neu_json_node_state_t *states = resp->states; + for (int i = 0; i < resp->n_state; i++) { + neu_json_elem_t state_elems[] = { { + .name = "node", + .t = NEU_JSON_STR, + .v.val_str = (&states[i])->node, + }, + { + .name = "link", + .t = NEU_JSON_INT, + .v.val_int = (&states[i])->link, + }, + { + .name = "running", + .t = NEU_JSON_INT, + .v.val_int = + (&states[i])->running, + } }; + state_array = neu_json_encode_array(state_array, state_elems, + NEU_JSON_ELEM_SIZE(state_elems)); + } + + neu_json_elem_t resp_elems[] = { { + .name = "states", + .t = NEU_JSON_OBJECT, + .v.val_object = state_array, + } }; + + ret = neu_json_encode_field(json_object, resp_elems, + NEU_JSON_ELEM_SIZE(resp_elems)); + + return ret; +} From 96992eff71e834384b25f9ca3c096d288b37b14d Mon Sep 17 00:00:00 2001 From: gc87 Date: Thu, 14 Jul 2022 16:32:22 +0800 Subject: [PATCH 6/9] refactor(mqtt): Modify unreasonable code * Modify header file ordering * Modify mqtt.* to mqtt_*.* to avoid ide parsing exception * Modify part of the code according to the pr revision comments --- include/neuron/json/neu_json_mqtt.h | 2 +- plugins/mqtt/CMakeLists.txt | 7 ++++--- plugins/mqtt/command/command.c | 5 +---- plugins/mqtt/command/common.h | 2 ++ plugins/mqtt/command/heartbeat.c | 19 ++++++++++--------- plugins/mqtt/command/read_write.c | 2 -- plugins/mqtt/{mqtt.c => mqtt_plugin.c} | 7 ++++--- plugins/mqtt/{mqtt.h => mqtt_plugin.h} | 7 ++++--- plugins/mqtt/{util.c => mqtt_util.c} | 5 ++--- plugins/mqtt/{util.h => mqtt_util.h} | 0 src/connection/mqtt_c_client.c | 1 - src/parser/neu_json_mqtt.c | 11 +++++------ tests/CMakeLists.txt | 2 +- tests/mqtt_test.cc | 2 +- 14 files changed, 35 insertions(+), 37 deletions(-) rename plugins/mqtt/{mqtt.c => mqtt_plugin.c} (99%) rename plugins/mqtt/{mqtt.h => mqtt_plugin.h} (95%) rename plugins/mqtt/{util.c => mqtt_util.c} (98%) rename plugins/mqtt/{util.h => mqtt_util.h} (100%) diff --git a/include/neuron/json/neu_json_mqtt.h b/include/neuron/json/neu_json_mqtt.h index c091b66d..16ebfc72 100644 --- a/include/neuron/json/neu_json_mqtt.h +++ b/include/neuron/json/neu_json_mqtt.h @@ -43,7 +43,7 @@ typedef struct { typedef struct { char * version; - uint32_t timpstamp; + uint64_t timpstamp; } neu_json_states_head_t; typedef struct { diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index 983c0145..d89dfd99 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -7,14 +7,15 @@ set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins") file(COPY ${CMAKE_SOURCE_DIR}/plugins/mqtt/mqtt.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/) set(src - mqtt.c - util.c + mqtt_plugin.c + mqtt_util.c command/read_write.c command/heartbeat.c command/command.c) add_library(${PROJECT_NAME} SHARED ${src}) target_include_directories(${PROJECT_NAME} PRIVATE - ${CMAKE_SOURCE_DIR}/include/neuron) + ${CMAKE_SOURCE_DIR}/include/neuron + ${CMAKE_SOURCE_DIR}/plugins/mqtt) target_link_libraries(${PROJECT_NAME} neuron-base) target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT}) diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index c39f6bf8..3f871bfd 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -19,12 +19,9 @@ #include #include +#include "command.h" #include "heartbeat.h" #include "read_write.h" -#include "utils/log.h" - -#include "../mqtt.h" -#include "command.h" // Read tags with group config static int read_request(mqtt_response_t *response, neu_plugin_t *plugin, diff --git a/plugins/mqtt/command/common.h b/plugins/mqtt/command/common.h index a306edb6..d1762e59 100644 --- a/plugins/mqtt/command/common.h +++ b/plugins/mqtt/command/common.h @@ -30,6 +30,8 @@ extern "C" { #include "json/neu_json_mqtt.h" #include "json/neu_json_rw.h" +#include "mqtt_plugin.h" + #define UNUSED(x) (void) (x) #ifdef __cplusplus diff --git a/plugins/mqtt/command/heartbeat.c b/plugins/mqtt/command/heartbeat.c index e60625fd..f668b990 100644 --- a/plugins/mqtt/command/heartbeat.c +++ b/plugins/mqtt/command/heartbeat.c @@ -21,11 +21,10 @@ #include #include -#include "../mqtt.h" -#include "heartbeat.h" -#include "utils/log.h" #include "version.h" +#include "heartbeat.h" + char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states) { char * version = NEURON_VERSION; @@ -38,12 +37,14 @@ char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states) json.states = calloc(json.n_state, sizeof(neu_json_node_state_t)); } - for (int i = 0; i < json.n_state; i++) { - struct node_state **nsp = utarray_eltptr(states, (unsigned int) i); - struct node_state * ns = (*nsp); - json.states[i].node = ns->node; - json.states[i].link = ns->link; - json.states[i].running = ns->running; + int index = 0; + utarray_foreach(states, struct node_state **, state_p) + { + struct node_state *ns = (*state_p); + json.states[index].node = ns->node; + json.states[index].link = ns->link; + json.states[index].running = ns->running; + index++; } neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header, diff --git a/plugins/mqtt/command/read_write.c b/plugins/mqtt/command/read_write.c index 0a8d2375..b251415d 100644 --- a/plugins/mqtt/command/read_write.c +++ b/plugins/mqtt/command/read_write.c @@ -21,9 +21,7 @@ #include #include -#include "../mqtt.h" #include "read_write.h" -#include "utils/log.h" // static uint64_t current_time() // { diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt_plugin.c similarity index 99% rename from plugins/mqtt/mqtt.c rename to plugins/mqtt/mqtt_plugin.c index 98faa837..55987e1f 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -18,12 +18,13 @@ **/ #include + #include #include #include "command/command.h" -#include "mqtt.h" -#include "util.h" +#include "mqtt_plugin.h" +#include "mqtt_util.h" #define INTERVAL 100000U #define TIMEOUT 3000U @@ -135,7 +136,7 @@ static void topics_generate(UT_array *topics, char *name, char *upload_topic, char *heartbeat_req = NULL; char *heartbeat_res = NULL; if (NULL != heartbeat_topic && 0 < strlen(heartbeat_topic)) { - nlog_debug("------------->%s", heartbeat_topic); + nlog_debug("user defined heartbeat topic:%s", heartbeat_topic); heartbeat_res = strdup(heartbeat_topic); } else { heartbeat_res = topics_format(TOPIC_HEARTBEAT_RES, name); diff --git a/plugins/mqtt/mqtt.h b/plugins/mqtt/mqtt_plugin.h similarity index 95% rename from plugins/mqtt/mqtt.h rename to plugins/mqtt/mqtt_plugin.h index 93818d30..9864e525 100644 --- a/plugins/mqtt/mqtt.h +++ b/plugins/mqtt/mqtt_plugin.h @@ -17,16 +17,17 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#ifndef NEURON_PLUGIN_MQTT -#define NEURON_PLUGIN_MQTT +#ifndef NEURON_MQTT_PLUGIN +#define NEURON_MQTT_PLUGIN #ifdef __cplusplus extern "C" { #endif -#include #include +#include + typedef struct mqtt_routine mqtt_routine_t; struct node_state { diff --git a/plugins/mqtt/util.c b/plugins/mqtt/mqtt_util.c similarity index 98% rename from plugins/mqtt/util.c rename to plugins/mqtt/mqtt_util.c index ca73d2e6..d655c37f 100644 --- a/plugins/mqtt/util.c +++ b/plugins/mqtt/mqtt_util.c @@ -17,16 +17,15 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#include #include #include #include #include -#include "util.h" #include #include -#include + +#include "mqtt_util.h" void mqtt_option_uninit(neu_mqtt_option_t *option) { diff --git a/plugins/mqtt/util.h b/plugins/mqtt/mqtt_util.h similarity index 100% rename from plugins/mqtt/util.h rename to plugins/mqtt/mqtt_util.h diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index 9955403a..4e478f2a 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -274,7 +274,6 @@ static BIO *bio_tcp_create(const char *addr, const char *port) return NULL; } - BIO_clear_flags(bio, BIO_FLAGS_SHOULD_RETRY); BIO_set_nbio(bio, 1); BIO_set_conn_port(bio, port); return bio; diff --git a/src/parser/neu_json_mqtt.c b/src/parser/neu_json_mqtt.c index 8f560997..8702c9e8 100644 --- a/src/parser/neu_json_mqtt.c +++ b/src/parser/neu_json_mqtt.c @@ -119,18 +119,17 @@ int neu_json_encode_states_resp(void *json_object, void *param) neu_json_elem_t state_elems[] = { { .name = "node", .t = NEU_JSON_STR, - .v.val_str = (&states[i])->node, + .v.val_str = states[i].node, }, { .name = "link", .t = NEU_JSON_INT, - .v.val_int = (&states[i])->link, + .v.val_int = states[i].link, }, { - .name = "running", - .t = NEU_JSON_INT, - .v.val_int = - (&states[i])->running, + .name = "running", + .t = NEU_JSON_INT, + .v.val_int = states[i].running, } }; state_array = neu_json_encode_array(state_array, state_elems, NEU_JSON_ELEM_SIZE(state_elems)); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index d00b6f7b..d64fe4a9 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -19,7 +19,7 @@ target_link_libraries(http_test neuron-base gtest_main gtest jansson nng) # file(COPY ${CMAKE_SOURCE_DIR}/plugin_param_schema.json DESTINATION ${CMAKE_BINARY_DIR}/tests) add_executable(mqtt_test mqtt_test.cc - ${CMAKE_SOURCE_DIR}/plugins/mqtt/util.c) + ${CMAKE_SOURCE_DIR}/plugins/mqtt/mqtt_util.c) target_include_directories(mqtt_test PRIVATE ${CMAKE_SOURCE_DIR}/src ${CMAKE_SOURCE_DIR}/include diff --git a/tests/mqtt_test.cc b/tests/mqtt_test.cc index 6e7a7ef8..5d27c0e7 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -1,6 +1,6 @@ #include -#include "util.h" +#include "mqtt_util.h" #include #include "utils/log.h" From 3c59046314cce3933ec2a06277ae60eef068bc96 Mon Sep 17 00:00:00 2001 From: gc87 Date: Thu, 14 Jul 2022 22:48:40 +0800 Subject: [PATCH 7/9] feat(mqtt): setting optional field --- plugins/mqtt/mqtt_plugin.c | 8 +- plugins/mqtt/mqtt_util.c | 152 ++++++++++++++++++++++++++------- plugins/mqtt/mqtt_util.h | 5 +- src/connection/mqtt_c_client.c | 85 +++++++++++------- tests/mqtt_test.cc | 12 ++- 5 files changed, 193 insertions(+), 69 deletions(-) diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 55987e1f..6f50c533 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -289,9 +289,9 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, } // MQTT client start - int rc = mqtt_option_init((char *) config, &routine->option); + int rc = mqtt_option_init(plugin, (char *) config, &routine->option); if (0 != rc) { - mqtt_option_uninit(&routine->option); + mqtt_option_uninit(plugin, &routine->option); free(routine); return NULL; } @@ -302,7 +302,7 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, neu_err_code_e error = NEU_ERR_SUCCESS; error = neu_mqtt_client_open(client_p, &routine->option, plugin); if (NEU_ERR_SUCCESS != error) { - mqtt_option_uninit(&routine->option); + mqtt_option_uninit(plugin, &routine->option); free(routine); return NULL; } @@ -362,7 +362,7 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) topics_cleanup(routine->topics); utarray_free(routine->topics); - mqtt_option_uninit(&routine->option); + mqtt_option_uninit(routine->plugin, &routine->option); } static neu_plugin_t *mqtt_plugin_open(void) diff --git a/plugins/mqtt/mqtt_util.c b/plugins/mqtt/mqtt_util.c index d655c37f..25fa93ae 100644 --- a/plugins/mqtt/mqtt_util.c +++ b/plugins/mqtt/mqtt_util.c @@ -25,10 +25,13 @@ #include #include +#include "mqtt_plugin.h" #include "mqtt_util.h" -void mqtt_option_uninit(neu_mqtt_option_t *option) +void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option) { + (void) plugin; + if (NULL != option->clientid) { free(option->clientid); option->clientid = NULL; @@ -90,7 +93,8 @@ void mqtt_option_uninit(neu_mqtt_option_t *option) } } -int mqtt_option_init(char *config, neu_mqtt_option_t *option) +int mqtt_option_init(neu_plugin_t *plugin, char *config, + neu_mqtt_option_t *option) { if (NULL == config || NULL == option) { return -1; @@ -113,45 +117,133 @@ int mqtt_option_init(char *config, neu_mqtt_option_t *option) neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR }; neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR }; - ret = neu_parse_param(config, &error, 13, &id, &upload, &heartbeat, &format, - &ssl, &host, &port, &username, &password, &ca, &cert, - &key, &keypass); + // client-id, required + ret = neu_parse_param(config, &error, 1, &id); if (0 != ret) { - return ret; - } - - if (false == ssl.v.val_bool) { - option->connection = strdup("tcp://"); + plog_error(plugin, "can't decode client-id from json"); + free(id.v.val_str); + return -1; } else { - option->connection = strdup("ssl://"); + option->clientid = id.v.val_str; } - if (0 == strcmp(option->connection, "ssl://")) { - if (0 == strlen(ca.v.val_str)) { - free(ca.v.val_str); - return -1; + // upload-topic, optional + ret = neu_parse_param(config, &error, 1, &upload); + if (0 != ret) { + free(upload.v.val_str); + } else { + option->upload_topic = upload.v.val_str; + } + + // heartbeat-topic, optional + ret = neu_parse_param(config, &error, 1, &heartbeat); + if (0 != ret) { + free(heartbeat.v.val_str); + } else { + option->heartbeat_topic = heartbeat.v.val_str; + } + + // format, required + ret = neu_parse_param(config, &error, 1, &format); + if (0 != ret) { + option->format = 0; + } else { + option->format = format.v.val_int; + } + + // ssl, required + ret = neu_parse_param(config, &error, 1, &ssl); + if (0 != ret) { + plog_error(plugin, "can't decode ssl from json"); + return -2; + } else { + if (false == ssl.v.val_bool) { + option->connection = strdup("tcp://"); + } else { + option->connection = strdup("ssl://"); + + // ca, required if ssl enabled + ret = neu_parse_param(config, &error, 1, &ca); + if (0 != ret) { + plog_error(plugin, "can't decode ca from json"); + free(ca.v.val_str); + return -3; + } else { + option->ca = ca.v.val_str; + } + + // cert, optional + ret = neu_parse_param(config, &error, 1, &cert); + if (0 == ret) { + free(cert.v.val_str); + } else { + option->cert = cert.v.val_str; + + // key, required if cert enable + ret = neu_parse_param(config, &error, 1, &key); + if (0 != ret) { + plog_error(plugin, "can't decode key from json"); + free(key.v.val_str); + return -4; + } else { + option->key = key.v.val_str; + } + + // keypass, optional + ret = neu_parse_param(config, &error, 1, &keypass); + if (0 != ret) { + free(keypass.v.val_str); + } else { + option->keypass = keypass.v.val_str; + } + } } } + // host, required + ret = neu_parse_param(config, &error, 1, &host); + if (0 != ret) { + plog_error(plugin, "can't decode host from json"); + free(host.v.val_str); + return -5; + } else { + option->host = host.v.val_str; + } + + // port, required + ret = neu_parse_param(config, &error, 1, &port); + if (0 != ret) { + plog_error(plugin, "can't decode port from json"); + return -6; + } else { + int32_t p = (int) port.v.val_int; + option->port = calloc(10, sizeof(char)); + snprintf(option->port, 10, "%d", p); + option->password = password.v.val_str; + } + + // username, optional + ret = neu_parse_param(config, &error, 1, &username); + if (0 != ret) { + free(username.v.val_str); + } else { + option->username = username.v.val_str; + } + + // password, optional + ret = neu_parse_param(config, &error, 1, &password); + if (0 != ret) { + free(password.v.val_str); + } else { + option->password = password.v.val_str; + } + // MQTT option - option->MQTT_version = 4; // Version 3.1.1 - option->host = host.v.val_str; - int32_t p = (int) port.v.val_int; - option->port = calloc(10, sizeof(char)); - snprintf(option->port, 10, "%d", p); - option->clientid = id.v.val_str; - option->upload_topic = upload.v.val_str; - option->heartbeat_topic = heartbeat.v.val_str; - option->format = format.v.val_int; - option->username = username.v.val_str; - option->password = password.v.val_str; - option->ca = ca.v.val_str; - option->cert = cert.v.val_str; - option->key = key.v.val_str; + option->MQTT_version = 4; // Version 3.1.1 option->keepalive_interval = 30; option->clean_session = 1; option->will_topic = NULL; option->will_payload = NULL; - option->keypass = keypass.v.val_str; + return 0; } diff --git a/plugins/mqtt/mqtt_util.h b/plugins/mqtt/mqtt_util.h index 983704d5..da86b081 100644 --- a/plugins/mqtt/mqtt_util.h +++ b/plugins/mqtt/mqtt_util.h @@ -27,8 +27,9 @@ extern "C" { #include "connection/mqtt_client_intf.h" #include -int mqtt_option_init(char *config, neu_mqtt_option_t *option); -void mqtt_option_uninit(neu_mqtt_option_t *option); +int mqtt_option_init(neu_plugin_t *plugin, char *config, + neu_mqtt_option_t *option); +void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option); #ifdef __cplusplus } diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index 4e478f2a..11fdb092 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -592,29 +592,48 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option, utarray_new(client->array, &tuple_icd); struct reconnect_state *state = &client->state; - state->hostname = strdup(client->option->host); - state->port = strdup(client->option->port); - state->client_id = strdup(client->option->clientid); - state->connection = strdup(client->option->connection); - state->username = strdup(client->option->username); - state->password = strdup(client->option->password); - state->keepalive = client->option->keepalive_interval; - state->clean = client->option->clean_session; - state->ca_file = strdup(client->option->ca); - state->cert_file = strdup(client->option->cert); - state->key_file = strdup(client->option->key); - state->client = client; - state->send_buf_sz = sizeof(state->send_buf); - state->recv_buf_sz = sizeof(state->recv_buf); - state->cert = NULL; - state->key = NULL; - state->ssl_ctx = NULL; - state->sock_fd = NULL; - state->running = true; - state->working = true; - pthread_mutex_init(&state->sock_state_mutex, NULL); - pthread_mutex_init(&state->running_mutex, NULL); - pthread_mutex_init(&state->working_mutex, NULL); + + if (NULL != client->option->clientid) { + state->client_id = strdup(client->option->clientid); + } + + if (NULL != client->option->connection) { + state->connection = strdup(client->option->connection); + } + + if (NULL != client->option->host) { + state->hostname = strdup(client->option->host); + } + + if (NULL != client->option->port) { + state->port = strdup(client->option->port); + } + + if (NULL != client->option->username) { + state->username = strdup(client->option->username); + } + + if (NULL != client->option->password) { + state->password = strdup(client->option->password); + } + + if (NULL != client->option->ca) { + state->ca_file = strdup(client->option->ca); + } + + if (NULL != client->option->cert) { + state->cert_file = strdup(client->option->cert); + } + + if (NULL != client->option->key) { + state->key_file = strdup(client->option->key); + } + + if (NULL != client->option->keypass) { + state->keypass = strdup(client->option->keypass); + } else { + state->keypass = NULL; + } if (NULL != client->option->will_topic) { state->will_topic = strdup(option->will_topic); @@ -624,12 +643,20 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option, } } - if (NULL != client->option->keypass && - 0 < strlen(client->option->keypass)) { - state->keypass = strdup(client->option->keypass); - } else { - state->keypass = NULL; - } + state->client = client; + state->send_buf_sz = sizeof(state->send_buf); + state->recv_buf_sz = sizeof(state->recv_buf); + state->cert = NULL; + state->key = NULL; + state->ssl_ctx = NULL; + state->sock_fd = NULL; + state->running = true; + state->working = true; + state->keepalive = client->option->keepalive_interval; + state->clean = client->option->clean_session; + pthread_mutex_init(&state->sock_state_mutex, NULL); + pthread_mutex_init(&state->running_mutex, NULL); + pthread_mutex_init(&state->working_mutex, NULL); client->mqtt.publish_response_callback_state = state; return client; diff --git a/tests/mqtt_test.cc b/tests/mqtt_test.cc index 5d27c0e7..7c7efef7 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -1,14 +1,17 @@ #include -#include "mqtt_util.h" #include -#include "utils/log.h" +#include "mqtt_plugin.h" +#include "mqtt_util.h" zlog_category_t *neuron = NULL; TEST(MQTTTest, mqtt_option_init_by_config) { + neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t)); + neu_plugin_common_init(&plugin->common); + neu_mqtt_option_t option; memset(&option, 0, sizeof(neu_mqtt_option_t)); char *config = @@ -20,9 +23,10 @@ TEST(MQTTTest, mqtt_option_init_by_config) "\"\",\"ca\":\"\", \"cert\":\"\", \"key\":\"\", " "\"keypass\":\"\"}}"; - int rc = mqtt_option_init(config, &option); + int rc = mqtt_option_init(plugin, config, &option); EXPECT_EQ(0, rc); - mqtt_option_uninit(&option); + mqtt_option_uninit(plugin, &option); + free(plugin); } int main(int argc, char **argv) From 094dc7f1222f4b420ce741829449dfc211056c97 Mon Sep 17 00:00:00 2001 From: gc87 Date: Fri, 15 Jul 2022 09:28:27 +0800 Subject: [PATCH 8/9] refactor(mqtt): nng timer -> event timer --- plugins/mqtt/mqtt_plugin.c | 178 ++++++++++++++++--------------------- 1 file changed, 79 insertions(+), 99 deletions(-) diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 6f50c533..578768d6 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -52,13 +52,14 @@ struct topic_pair { }; struct mqtt_routine { - neu_plugin_t * plugin; - neu_mqtt_option_t option; - neu_mqtt_client_t client; - UT_array * topics; - nng_aio * aio; - nng_mtx * mutex; - UT_array * states; + neu_plugin_t * plugin; + neu_mqtt_option_t option; + neu_mqtt_client_t client; + UT_array * topics; + nng_mtx * mutex; + UT_array * states; + neu_events_t * events; + neu_event_timer_t *send; }; static neu_plugin_t *plugin_log = NULL; @@ -219,62 +220,52 @@ static void mqtt_routine_state(void *context, int state) } } -static void mqtt_routine_heartbeat(nng_aio *aio, void *arg, int code) +static int mqtt_routine_heartbeat(void *data) { - mqtt_routine_t *routine = (mqtt_routine_t *) arg; + mqtt_routine_t *routine = (mqtt_routine_t *) data; - switch (code) { - case NNG_ETIMEDOUT: { - // command all node list - neu_req_get_node_t cmd = { 0 }; - neu_reqresp_head_t header = { - .ctx = NULL, - .type = NEU_REQ_GET_NODE, - }; + neu_req_get_node_t cmd = { 0 }; + neu_reqresp_head_t header = { + .ctx = NULL, + .type = NEU_REQ_GET_NODE, + }; - cmd.type = NEU_NA_TYPE_APP; - int ret = neu_plugin_op(routine->plugin, header, &cmd); - if (ret != 0) { } - cmd.type = NEU_NA_TYPE_DRIVER; - ret = neu_plugin_op(routine->plugin, header, &cmd); - if (ret != 0) { } + cmd.type = NEU_NA_TYPE_APP; + int ret = neu_plugin_op(routine->plugin, header, &cmd); + if (ret != 0) { + return -1; + } + cmd.type = NEU_NA_TYPE_DRIVER; + ret = neu_plugin_op(routine->plugin, header, &cmd); + if (ret != 0) { + return -1; + } - nng_aio_set_timeout(aio, 3000); - nng_aio_defer(aio, mqtt_routine_heartbeat, routine); - - // publish state data - nng_mtx_lock(routine->mutex); - int type = TOPIC_TYPE_HEARTBEAT; - struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json_str = - command_heartbeat_response(routine->plugin, routine->states); - if (NULL == json_str) { - nng_mtx_unlock(routine->mutex); - break; - } - - 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) { - plog_error(plugin_log, - "heartbeat publish error code :%d, topoic:%s", error, - topic); - } - - free(json_str); + // publish state data + nng_mtx_lock(routine->mutex); + int type = TOPIC_TYPE_HEARTBEAT; + struct topic_pair *pair = topics_find_type(routine->topics, type); + char * json_str = + command_heartbeat_response(routine->plugin, routine->states); + if (NULL == json_str) { nng_mtx_unlock(routine->mutex); - break; + return -1; } - case NNG_ECANCELED: { - plog_info(plugin_log, "aio:%p, cancel", aio); - break; - } - default: - plog_warn(plugin_log, "aio:%p, skip error:%d", aio, code); + + 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) { + plog_error(plugin_log, "heartbeat publish error code :%d, topoic:%s", + error, topic); } + + free(json_str); + nng_mtx_unlock(routine->mutex); + + return 0; } static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, @@ -325,10 +316,15 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, // start heartbeat send nng_mtx_alloc(&routine->mutex); - nng_aio_alloc(&routine->aio, NULL, routine); - nng_aio_wait(routine->aio); - nng_aio_set_timeout(routine->aio, 3000); - nng_aio_defer(routine->aio, mqtt_routine_heartbeat, routine); + routine->events = neu_event_new(); + neu_event_timer_param_t param = { + .second = 3, + .millisecond = 0, + .usr_data = routine, + .cb = mqtt_routine_heartbeat, + }; + + routine->send = neu_event_add_timer(routine->events, param); return routine; } @@ -337,6 +333,11 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) { assert(NULL != routine); + // stop heartbeat send + neu_event_del_timer(routine->events, routine->send); + neu_event_close(routine->events); + nng_mtx_free(routine->mutex); + // cleanup node states for (struct node_state **sp = (struct node_state **) utarray_front(routine->states); @@ -347,11 +348,6 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) utarray_free(routine->states); - // stop heartbeat send - nng_mtx_free(routine->mutex); - nng_aio_cancel(routine->aio); - nng_aio_free(routine->aio); - // stop recevied neu_mqtt_client_suspend(routine->client); @@ -592,32 +588,21 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) return NEU_ERR_SUCCESS; } -// static neu_err_code_e reconnect(neu_plugin_t *plugin, neu_reqresp_head_t -// *head) -// { -// if (NEU_RESP_ERROR == head->type || NEU_RESP_READ_GROUP == head->type -// || -// NEU_REQRESP_TRANS_DATA == head->type) { -// if (!plugin->running && NULL != plugin->config) { -// mqtt_routine_t *routine = -// mqtt_routine_start(plugin, plugin->config); -// if (NULL == routine) { -// plugin->routine = NULL; -// plugin->running = false; -// return NEU_ERR_FAILURE; -// } +static neu_err_code_e subscribe_handle(neu_plugin_t * plugin, + neu_reqresp_head_t *head, void *data) +{ + UNUSED(plugin); + UNUSED(head); -// plugin->routine = routine; -// plugin->running = true; -// plugin->common.link_state = NEU_PLUGIN_LINK_STATE_CONNECTING; -// } -// } + neu_resp_app_subscribe_group_t *sub = + (neu_resp_app_subscribe_group_t *) data; + utarray_free(sub->tags); -// return NEU_ERR_SUCCESS; -// } + return NEU_ERR_SUCCESS; +} -static neu_err_code_e node_list_get(neu_plugin_t * plugin, - neu_reqresp_head_t *head, void *data) +static neu_err_code_e nodes_get(neu_plugin_t *plugin, neu_reqresp_head_t *head, + void *data) { UNUSED(head); @@ -632,12 +617,14 @@ static neu_err_code_e node_list_get(neu_plugin_t * plugin, strcpy(cmd.node, info->node); int ret = neu_plugin_op(plugin, header, &cmd); - if (ret != 0) { } + if (ret != 0) { + return -1; + } } utarray_free(nodes->nodes); - return NEU_ERR_SUCCESS; + return 0; } static neu_err_code_e node_state_update(neu_plugin_t * plugin, @@ -682,10 +669,6 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, assert(NULL != data); neu_err_code_e error = NEU_ERR_SUCCESS; - // error = reconnect(plugin, head); - // if (NEU_ERR_SUCCESS != error) { - // return error; - // } switch (head->type) { case NEU_RESP_ERROR: @@ -698,14 +681,11 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, error = trans_data(plugin, data); break; case NEU_RESP_APP_SUBSCRIBE_GROUP: { - neu_resp_app_subscribe_group_t *sub = - (neu_resp_app_subscribe_group_t *) data; - - utarray_free(sub->tags); + error = subscribe_handle(plugin, head, data); break; } case NEU_RESP_GET_NODE: { - error = node_list_get(plugin, head, data); + error = nodes_get(plugin, head, data); break; } case NEU_RESP_GET_NODE_STATE: { From 9d3b411627280f42887909669c3904056887ff15 Mon Sep 17 00:00:00 2001 From: gc87 Date: Fri, 15 Jul 2022 13:48:57 +0800 Subject: [PATCH 9/9] fix(mqtt): setting free * Add unit test for setting * Set mqtt.json format field to optinal --- plugins/mqtt/command/heartbeat.c | 8 +- plugins/mqtt/mqtt.json | 2 +- plugins/mqtt/mqtt_plugin.c | 4 +- plugins/mqtt/mqtt_util.c | 24 +++-- tests/mqtt_test.cc | 157 +++++++++++++++++++++++++++++-- 5 files changed, 169 insertions(+), 26 deletions(-) diff --git a/plugins/mqtt/command/heartbeat.c b/plugins/mqtt/command/heartbeat.c index f668b990..83c63281 100644 --- a/plugins/mqtt/command/heartbeat.c +++ b/plugins/mqtt/command/heartbeat.c @@ -33,13 +33,11 @@ char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states) neu_json_states_t json = { 0 }; char * json_str = NULL; json.n_state = utarray_len(states); - if (0 < json.n_state) { - json.states = calloc(json.n_state, sizeof(neu_json_node_state_t)); - } + json.states = calloc(json.n_state, sizeof(neu_json_node_state_t)); - int index = 0; utarray_foreach(states, struct node_state **, state_p) { + int index = utarray_eltidx(states, state_p); struct node_state *ns = (*state_p); json.states[index].node = ns->node; json.states[index].link = ns->link; @@ -49,7 +47,7 @@ char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states) neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header, neu_json_encode_state_header_resp, &json_str); - if (0 < json.n_state) { + if (NULL != json.states) { free(json.states); } diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index a454ede0..bb5997de 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -32,7 +32,7 @@ "format": { "name": "upload format", "description": "Upload format on mqtt", - "attribute": "required", + "attribute": "optional", "type": "map", "default": 0, "valid": { diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 578768d6..9a8b5b27 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -465,7 +465,7 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config) plugin_stop_running(plugin); int rc = plguin_start_running(plugin); if (0 != rc) { - return NEU_ERR_MQTT_FAILURE; + return NEU_ERR_NODE_SETTING_INVALID; } plog_info(plugin, "config plugin: %s", neu_plugin_module.module_name); @@ -478,7 +478,7 @@ static int mqtt_plugin_start(neu_plugin_t *plugin) int rc = plguin_start_running(plugin); if (0 != rc) { - return NEU_ERR_MQTT_FAILURE; + return NEU_ERR_NODE_SETTING_INVALID; } return NEU_ERR_SUCCESS; diff --git a/plugins/mqtt/mqtt_util.c b/plugins/mqtt/mqtt_util.c index 25fa93ae..673afb6a 100644 --- a/plugins/mqtt/mqtt_util.c +++ b/plugins/mqtt/mqtt_util.c @@ -96,6 +96,7 @@ void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option) int mqtt_option_init(neu_plugin_t *plugin, char *config, neu_mqtt_option_t *option) { + (void) plugin; if (NULL == config || NULL == option) { return -1; } @@ -120,7 +121,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // client-id, required ret = neu_parse_param(config, &error, 1, &id); if (0 != ret) { - plog_error(plugin, "can't decode client-id from json"); + free(error); free(id.v.val_str); return -1; } else { @@ -130,6 +131,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // upload-topic, optional ret = neu_parse_param(config, &error, 1, &upload); if (0 != ret) { + free(error); free(upload.v.val_str); } else { option->upload_topic = upload.v.val_str; @@ -138,14 +140,16 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // heartbeat-topic, optional ret = neu_parse_param(config, &error, 1, &heartbeat); if (0 != ret) { + free(error); free(heartbeat.v.val_str); } else { option->heartbeat_topic = heartbeat.v.val_str; } - // format, required + // format, optional ret = neu_parse_param(config, &error, 1, &format); if (0 != ret) { + free(error); option->format = 0; } else { option->format = format.v.val_int; @@ -154,7 +158,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // ssl, required ret = neu_parse_param(config, &error, 1, &ssl); if (0 != ret) { - plog_error(plugin, "can't decode ssl from json"); + free(error); return -2; } else { if (false == ssl.v.val_bool) { @@ -165,7 +169,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // ca, required if ssl enabled ret = neu_parse_param(config, &error, 1, &ca); if (0 != ret) { - plog_error(plugin, "can't decode ca from json"); + free(error); free(ca.v.val_str); return -3; } else { @@ -174,7 +178,8 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // cert, optional ret = neu_parse_param(config, &error, 1, &cert); - if (0 == ret) { + if (0 != ret) { + free(error); free(cert.v.val_str); } else { option->cert = cert.v.val_str; @@ -182,7 +187,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // key, required if cert enable ret = neu_parse_param(config, &error, 1, &key); if (0 != ret) { - plog_error(plugin, "can't decode key from json"); + free(error); free(key.v.val_str); return -4; } else { @@ -192,6 +197,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // keypass, optional ret = neu_parse_param(config, &error, 1, &keypass); if (0 != ret) { + free(error); free(keypass.v.val_str); } else { option->keypass = keypass.v.val_str; @@ -203,7 +209,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // host, required ret = neu_parse_param(config, &error, 1, &host); if (0 != ret) { - plog_error(plugin, "can't decode host from json"); + free(error); free(host.v.val_str); return -5; } else { @@ -213,7 +219,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // port, required ret = neu_parse_param(config, &error, 1, &port); if (0 != ret) { - plog_error(plugin, "can't decode port from json"); + free(error); return -6; } else { int32_t p = (int) port.v.val_int; @@ -225,6 +231,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // username, optional ret = neu_parse_param(config, &error, 1, &username); if (0 != ret) { + free(error); free(username.v.val_str); } else { option->username = username.v.val_str; @@ -233,6 +240,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, // password, optional ret = neu_parse_param(config, &error, 1, &password); if (0 != ret) { + free(error); free(password.v.val_str); } else { option->password = password.v.val_str; diff --git a/tests/mqtt_test.cc b/tests/mqtt_test.cc index 7c7efef7..0f3d8079 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -13,19 +13,156 @@ TEST(MQTTTest, mqtt_option_init_by_config) neu_plugin_common_init(&plugin->common); neu_mqtt_option_t option; - memset(&option, 0, sizeof(neu_mqtt_option_t)); - char *config = - (char *) "{\"node_id\":5,\"params\":{\"client-id\":" - "\"upload123\", \"upload-topic\":\"\", " - "\"heartbeat-topic\":\"\", \"format\": " - "0,\"ssl\":false,\"host\":\"192.168.10." - "116\",\"port\":1883,\"username\":\"\",\"password\":" - "\"\",\"ca\":\"\", \"cert\":\"\", \"key\":\"\", " - "\"keypass\":\"\"}}"; + // memset(&option, 0, sizeof(neu_mqtt_option_t)); + // char *config = + // (char *) "{\"node_id\":5,\"params\":{\"client-id\":" + // "\"upload123\", \"upload-topic\":\"\", " + // "\"heartbeat-topic\":\"\", \"format\": " + // "0,\"ssl\":false,\"host\":\"192.168.10." + // "116\",\"port\":1883,\"username\":\"\",\"password\":" + // "\"\",\"ca\":\"\", \"cert\":\"\", \"key\":\"\", " + // "\"keypass\":\"\"}}"; - int rc = mqtt_option_init(plugin, config, &option); + // int rc = mqtt_option_init(plugin, config, &option); + // EXPECT_EQ(0, rc); + // mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config0 = (char *) "{\"node_id\":5,\"params\":{}}"; + int rc = mqtt_option_init(plugin, config0, &option); + EXPECT_EQ(-1, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config1 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\"}}"; + rc = mqtt_option_init(plugin, config1, &option); + EXPECT_EQ(-2, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config2 = (char *) "{\"node_id\":5,\"params\":{\"client-id\":" + "\"upload123\", \"upload-topic\":\"\"}}"; + rc = mqtt_option_init(plugin, config2, &option); + EXPECT_EQ(-2, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config3 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\"}}"; + rc = mqtt_option_init(plugin, config3, &option); + EXPECT_EQ(-2, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config4 = (char *) "{\"node_id\":5,\"params\":{\"client-id\":" + "\"upload123\", \"upload-topic\":\"\", " + "\"heartbeat-topic\":\"\", \"format\": 0}}"; + rc = mqtt_option_init(plugin, config4, &option); + EXPECT_EQ(-2, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config5 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":false}}"; + rc = mqtt_option_init(plugin, config5, &option); + EXPECT_EQ(-5, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config6 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":false,\"host\":\"192.168.10.116\"}}"; + rc = mqtt_option_init(plugin, config6, &option); + EXPECT_EQ(-6, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config7 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":false,\"host\":\"192.168.10.116\",\"port\":1883}}"; + rc = mqtt_option_init(plugin, config7, &option); EXPECT_EQ(0, rc); mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config8 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":false,\"host\":\"192.168.10.116\",\"port\":1883," + "\"username\":\"\"}}"; + rc = mqtt_option_init(plugin, config8, &option); + EXPECT_EQ(0, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config9 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":false,\"host\":\"192.168.10.116\",\"port\":1883," + "\"username\":\"\",\"password\":\"\"}}"; + rc = mqtt_option_init(plugin, config9, &option); + EXPECT_EQ(0, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config10 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":true,\"host\":\"192.168.10.116\",\"port\":1883," + "\"username\":\"\",\"password\":\"\"}}"; + rc = mqtt_option_init(plugin, config10, &option); + EXPECT_EQ(-3, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config11 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":true,\"host\":\"192.168.10.116\",\"port\":1883," + "\"username\":\"\",\"password\":\"\",\"ca\":\"\"}}"; + rc = mqtt_option_init(plugin, config11, &option); + EXPECT_EQ(0, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config12 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":true,\"host\":\"192.168.10.116\",\"port\":1883," + "\"username\":\"\",\"password\":\"\",\"ca\":\"\", " + "\"cert\":\"\"}}"; + rc = mqtt_option_init(plugin, config12, &option); + EXPECT_EQ(-4, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config13 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":false,\"host\":\"192.168.10.116\",\"port\":1883," + "\"username\":\"\",\"password\":\"\",\"ca\":\"\", " + "\"cert\":\"\", \"key\":\"\"}}"; + rc = mqtt_option_init(plugin, config13, &option); + EXPECT_EQ(0, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + char *config14 = + (char *) "{\"node_id\":5,\"params\":{\"client-id\":\"upload123\", " + "\"upload-topic\":\"\", \"heartbeat-topic\":\"\", \"format\": " + "0,\"ssl\":true,\"host\":\"192.168.10.116\",\"port\":1883," + "\"username\":\"\",\"password\":\"\",\"ca\":\"\", " + "\"cert\":\"\", \"key\":\"\", \"keypass\":\"\"}}"; + rc = mqtt_option_init(plugin, config14, &option); + EXPECT_EQ(0, rc); + mqtt_option_uninit(plugin, &option); + free(plugin); }