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/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/include/neuron/json/neu_json_mqtt.h b/include/neuron/json/neu_json_mqtt.h index a751ddae..16ebfc72 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; + uint64_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/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index 8549d8c1..d89dfd99 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -7,13 +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 3dffbbfa..3f871bfd 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -19,11 +19,9 @@ #include #include -#include "read_write.h" -#include "utils/log.h" - -#include "../mqtt.h" #include "command.h" +#include "heartbeat.h" +#include "read_write.h" // Read tags with group config static int read_request(mqtt_response_t *response, neu_plugin_t *plugin, @@ -128,4 +126,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, UT_array *states) +{ + 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 dd2cb9c8..be19d464 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 { @@ -56,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, UT_array *states); + #ifdef __cplusplus } #endif 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 new file mode 100644 index 00000000..83c63281 --- /dev/null +++ b/plugins/mqtt/command/heartbeat.c @@ -0,0 +1,55 @@ +/** + * 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 "version.h" + +#include "heartbeat.h" + +char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states) +{ + 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); + json.states = calloc(json.n_state, sizeof(neu_json_node_state_t)); + + 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; + json.states[index].running = ns->running; + index++; + } + + neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header, + neu_json_encode_state_header_resp, &json_str); + if (NULL != json.states) { + 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 new file mode 100644 index 00000000..5009a0c0 --- /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, UT_array *states); + +#ifdef __cplusplus +} +#endif +#endif 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.json b/plugins/mqtt/mqtt.json index 869d8d16..bb5997de 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -19,10 +19,20 @@ "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", - "attribute": "required", + "attribute": "optional", "type": "map", "default": 0, "valid": { diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt_plugin.c similarity index 70% rename from plugins/mqtt/mqtt.c rename to plugins/mqtt/mqtt_plugin.c index 31d64344..9a8b5b27 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -19,9 +19,12 @@ #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 @@ -32,6 +35,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 @@ -48,10 +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; + 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; @@ -103,7 +111,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 +132,19 @@ 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)) { + nlog_debug("user defined heartbeat topic:%s", 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) @@ -198,6 +220,54 @@ static void mqtt_routine_state(void *context, int state) } } +static int mqtt_routine_heartbeat(void *data) +{ + mqtt_routine_t *routine = (mqtt_routine_t *) data; + + 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) { + return -1; + } + cmd.type = NEU_NA_TYPE_DRIVER; + ret = neu_plugin_op(routine->plugin, header, &cmd); + if (ret != 0) { + return -1; + } + + // 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); + return -1; + } + + 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, const char * config) { @@ -210,20 +280,20 @@ 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; } 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); + mqtt_option_uninit(plugin, &routine->option); free(routine); return NULL; } @@ -233,12 +303,29 @@ 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.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_mtx_alloc(&routine->mutex); + 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; } @@ -246,6 +333,21 @@ 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); + sp != NULL; + sp = (struct node_state **) utarray_next(routine->states, sp)) { + free((*sp)); + } + + utarray_free(routine->states); + // stop recevied neu_mqtt_client_suspend(routine->client); @@ -256,7 +358,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) @@ -363,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); @@ -376,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; @@ -486,29 +588,78 @@ 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 nodes_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) { + return -1; + } + } + + utarray_free(nodes->nodes); + + return 0; +} + +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; + + 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)) { + + 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; + nng_mtx_unlock(routine->mutex); + + free(head->ctx); + + return NEU_ERR_SUCCESS; +} static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, void *data) @@ -518,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: @@ -534,10 +681,15 @@ 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 = nodes_get(plugin, head, data); + break; + } + case NEU_RESP_GET_NODE_STATE: { + error = node_state_update(plugin, head, data); break; } default: diff --git a/plugins/mqtt/mqtt.h b/plugins/mqtt/mqtt_plugin.h similarity index 86% rename from plugins/mqtt/mqtt.h rename to plugins/mqtt/mqtt_plugin.h index 6bac6bb5..9864e525 100644 --- a/plugins/mqtt/mqtt.h +++ b/plugins/mqtt/mqtt_plugin.h @@ -17,18 +17,26 @@ * 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 { + 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/mqtt_util.c b/plugins/mqtt/mqtt_util.c new file mode 100644 index 00000000..673afb6a --- /dev/null +++ b/plugins/mqtt/mqtt_util.c @@ -0,0 +1,257 @@ +/** + * 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 + +#include +#include + +#include "mqtt_plugin.h" +#include "mqtt_util.h" + +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; + } + + if (NULL != option->upload_topic) { + free(option->upload_topic); + 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; + } + + if (NULL != option->host) { + free(option->host); + option->host = NULL; + } + + if (NULL != option->port) { + free(option->port); + option->port = NULL; + } + + if (NULL != option->username) { + free(option->username); + option->username = NULL; + } + + if (NULL != option->password) { + free(option->password); + option->password = NULL; + } + + if (NULL != option->ca) { + free(option->ca); + option->ca = NULL; + } + + if (NULL != option->cert) { + free(option->cert); + option->cert = NULL; + } + + if (NULL != option->key) { + free(option->key); + option->key = NULL; + } + + if (NULL != option->keypass) { + free(option->keypass); + option->keypass = NULL; + } +} + +int mqtt_option_init(neu_plugin_t *plugin, char *config, + neu_mqtt_option_t *option) +{ + (void) plugin; + if (NULL == config || NULL == 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 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 }; + + // client-id, required + ret = neu_parse_param(config, &error, 1, &id); + if (0 != ret) { + free(error); + free(id.v.val_str); + return -1; + } else { + option->clientid = id.v.val_str; + } + + // 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; + } + + // 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, optional + ret = neu_parse_param(config, &error, 1, &format); + if (0 != ret) { + free(error); + option->format = 0; + } else { + option->format = format.v.val_int; + } + + // ssl, required + ret = neu_parse_param(config, &error, 1, &ssl); + if (0 != ret) { + free(error); + 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) { + free(error); + 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(error); + 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) { + free(error); + 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(error); + 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) { + free(error); + 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) { + free(error); + 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(error); + 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(error); + free(password.v.val_str); + } else { + option->password = password.v.val_str; + } + + // MQTT option + option->MQTT_version = 4; // Version 3.1.1 + option->keepalive_interval = 30; + option->clean_session = 1; + option->will_topic = NULL; + option->will_payload = NULL; + + return 0; +} diff --git a/plugins/mqtt/util.h b/plugins/mqtt/mqtt_util.h similarity index 85% rename from plugins/mqtt/util.h rename to plugins/mqtt/mqtt_util.h index 983704d5..da86b081 100644 --- a/plugins/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/plugins/mqtt/util.c b/plugins/mqtt/util.c deleted file mode 100644 index b66ab0a5..00000000 --- a/plugins/mqtt/util.c +++ /dev/null @@ -1,150 +0,0 @@ -/** - * 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 -#include - -#include "util.h" -#include -#include -#include - -void mqtt_option_uninit(neu_mqtt_option_t *option) -{ - if (NULL != option->clientid) { - free(option->clientid); - option->clientid = NULL; - } - - if (NULL != option->upload_topic) { - free(option->upload_topic); - option->upload_topic = NULL; - } - - if (NULL != option->connection) { - free(option->connection); - option->connection = NULL; - } - - if (NULL != option->host) { - free(option->host); - option->host = NULL; - } - - if (NULL != option->port) { - free(option->port); - option->port = NULL; - } - - if (NULL != option->username) { - free(option->username); - option->username = NULL; - } - - if (NULL != option->password) { - free(option->password); - option->password = NULL; - } - - if (NULL != option->ca) { - free(option->ca); - option->ca = NULL; - } - - if (NULL != option->cert) { - free(option->cert); - option->cert = NULL; - } - - if (NULL != option->key) { - free(option->key); - option->key = NULL; - } - - if (NULL != option->keypass) { - free(option->keypass); - option->keypass = NULL; - } -} - -int mqtt_option_init(char *config, neu_mqtt_option_t *option) -{ - if (NULL == config || NULL == 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 }; - - ret = neu_parse_param(config, &error, 12, &id, &upload, &format, &ssl, - &host, &port, &username, &password, &ca, &cert, &key, - &keypass); - if (0 != ret) { - return ret; - } - - if (false == ssl.v.val_bool) { - option->connection = strdup("tcp://"); - } else { - option->connection = strdup("ssl://"); - } - - if (0 == strcmp(option->connection, "ssl://")) { - if (0 == strlen(ca.v.val_str)) { - free(ca.v.val_str); - return -1; - } - } - - // 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->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->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/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/src/parser/neu_json_mqtt.c b/src/parser/neu_json_mqtt.c index 0d3d0c14..8702c9e8 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,63 @@ 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; +} 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 cc242ae7..0f3d8079 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -1,27 +1,169 @@ #include -#include "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_mqtt_option_t option; - memset(&option, 0, sizeof(neu_mqtt_option_t)); - char *config = - (char *) "{\"node_id\":5,\"params\":{\"client-id\":" - "\"upload123\", \"upload-topic\":\"\", \"format\": " - "0,\"ssl\":false,\"host\":\"192.168.10." - "116\",\"port\":1883,\"username\":\"\",\"password\":" - "\"\",\"ca\":\"\", \"cert\":\"\", \"key\":\"\", " - "\"keypass\":\"\"}}"; + neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t)); + neu_plugin_common_init(&plugin->common); - int rc = mqtt_option_init(config, &option); + 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\":\"\"}}"; + + // 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(&option); + 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); } int main(int argc, char **argv)