From caff948cd6fc9bd070d4f76e2674a7dec367a9e4 Mon Sep 17 00:00:00 2001 From: gc87 Date: Wed, 24 Aug 2022 17:50:02 +0800 Subject: [PATCH] refactor: rm clientid & make topic field required --- plugins/mqtt/mqtt.json | 18 +--- plugins/mqtt/mqtt_plugin.c | 15 +-- plugins/mqtt/mqtt_util.c | 94 ++++++++++++++--- plugins/mqtt/mqtt_util.h | 7 +- tests/mqtt_test.cc | 200 +++++++++++++++++++------------------ 5 files changed, 196 insertions(+), 138 deletions(-) diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index bb5997de..2c3d2a2f 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -1,20 +1,10 @@ { - "client-id": { - "name": "client id", - "description": "Neuron MQTT client ID", - "type": "string", - "attribute": "required", - "default": "", - "valid": { - "length": 255 - } - }, "upload-topic": { "name": "upload topic", "description": "User-defined upload topic", "type": "string", - "attribute": "optional", - "default": "", + "attribute": "required", + "default": "/neuron/${node-name}/upload", "valid": { "length": 255 } @@ -23,8 +13,8 @@ "name": "heartbeat topic", "description": "User-defined heartbeat topic", "type": "string", - "attribute": "optional", - "default": "", + "attribute": "required", + "default": "/neuron/${node-name}/heartbeat", "valid": { "length": 255 } diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 68b8eef6..3f958d93 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -30,13 +30,13 @@ #define TIMEOUT 3000U #define MAX_TOPIC_LEN 256 -#define TOPIC_READ_REQ "neuron/%s/read/req" -#define TOPIC_WRITE_REQ "neuron/%s/write/req" +#define TOPIC_READ_REQ "/neuron/%s/read/req" +#define TOPIC_WRITE_REQ "/neuron/%s/write/req" -#define TOPIC_READ_RES "neuron/%s/read/resp" -#define TOPIC_WRITE_RES "neuron/%s/write/resp" -#define TOPIC_UPLOAD_RES "neuron/%s/upload" -#define TOPIC_HEARTBEAT_RES "neuron/%s/heartbeat" +#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 @@ -117,7 +117,7 @@ static void topics_cleanup(UT_array *topics, neu_plugin_t *plugin) static void topics_generate(UT_array *topics, neu_plugin_t *plugin, neu_mqtt_option_t *option) { - char *name = option->clientid; + char *name = plugin->common.name; char *upload_topic = option->upload_topic; char *heartbeat_topic = option->heartbeat_topic; @@ -244,6 +244,7 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, // MQTT client start int rc = mqtt_option_init(plugin, (char *) config, &routine->option); if (0 != rc) { + plog_error(plugin, "%s", mqtt_option_error(rc)); mqtt_option_uninit(plugin, &routine->option); free(routine); return NULL; diff --git a/plugins/mqtt/mqtt_util.c b/plugins/mqtt/mqtt_util.c index 673afb6a..cf9ceb04 100644 --- a/plugins/mqtt/mqtt_util.c +++ b/plugins/mqtt/mqtt_util.c @@ -22,12 +22,41 @@ #include #include +#include + #include #include #include "mqtt_plugin.h" #include "mqtt_util.h" +int uuid_v4(char *buffer) +{ + union { + struct { + uint32_t time_low; + uint16_t time_mid; + uint16_t time_hi_and_version; + uint8_t clk_seq_hi_res; + uint8_t clk_seq_low; + uint8_t node[6]; + }; + uint8_t __rnd[16]; + } uuid; + + int rc = RAND_bytes(uuid.__rnd, sizeof(uuid)); + uuid.clk_seq_hi_res = (uint8_t)((uuid.clk_seq_hi_res & 0x3F) | 0x80); + uuid.time_hi_and_version = + (uint16_t)((uuid.time_hi_and_version & 0x0FFF) | 0x4000); + + snprintf(buffer, 38, "%08x-%04x-%04x-%02x%02x-%02x%02x%02x%02x%02x%02x", + uuid.time_low, uuid.time_mid, uuid.time_hi_and_version, + uuid.clk_seq_hi_res, uuid.clk_seq_low, uuid.node[0], uuid.node[1], + uuid.node[2], uuid.node[3], uuid.node[4], uuid.node[5]); + + return rc; +} + void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option) { (void) plugin; @@ -98,12 +127,11 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, { (void) plugin; if (NULL == config || NULL == option) { - return -1; + 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 }; @@ -119,30 +147,41 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, 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; + char uuid[38] = { 0 }; + int rc = uuid_v4(uuid); + if (1 != rc) { + return 2; } - // upload-topic, optional + option->clientid = strdup(uuid); + + // upload-topic, required ret = neu_parse_param(config, &error, 1, &upload); if (0 != ret) { free(error); free(upload.v.val_str); + return 3; } else { + if (0 == strlen(upload.v.val_str)) { + free(upload.v.val_str); + return 3; + } + option->upload_topic = upload.v.val_str; } - // heartbeat-topic, optional + // heartbeat-topic, required ret = neu_parse_param(config, &error, 1, &heartbeat); if (0 != ret) { free(error); free(heartbeat.v.val_str); + return 4; } else { + if (0 == strlen(heartbeat.v.val_str)) { + free(heartbeat.v.val_str); + return 4; + } + option->heartbeat_topic = heartbeat.v.val_str; } @@ -159,7 +198,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, ret = neu_parse_param(config, &error, 1, &ssl); if (0 != ret) { free(error); - return -2; + return 5; } else { if (false == ssl.v.val_bool) { option->connection = strdup("tcp://"); @@ -171,7 +210,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, if (0 != ret) { free(error); free(ca.v.val_str); - return -3; + return 6; } else { option->ca = ca.v.val_str; } @@ -189,7 +228,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, if (0 != ret) { free(error); free(key.v.val_str); - return -4; + return 7; } else { option->key = key.v.val_str; } @@ -211,7 +250,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, if (0 != ret) { free(error); free(host.v.val_str); - return -5; + return 8; } else { option->host = host.v.val_str; } @@ -220,7 +259,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, ret = neu_parse_param(config, &error, 1, &port); if (0 != ret) { free(error); - return -6; + return 9; } else { int32_t p = (int) port.v.val_int; option->port = calloc(10, sizeof(char)); @@ -255,3 +294,26 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, return 0; } + +static const char *mqtt_option_error_strs[] = { + [0] = "success", + [1] = "config or option is null", + [2] = "generate uuid failed", + [3] = "upload topic field not set", + [4] = "heartbeat topic field not set", + [5] = "ssl field not set", + [6] = "ca field not set when login with tls", + [7] = "key field not set when login with tls with cert", + [8] = "host field not set", + [9] = "port field not set", + [10] = "unknow", +}; + +const char *mqtt_option_error(int error) +{ + if (0 <= error && 9 >= error) { + return mqtt_option_error_strs[error]; + } else { + return mqtt_option_error_strs[10]; + } +} \ No newline at end of file diff --git a/plugins/mqtt/mqtt_util.h b/plugins/mqtt/mqtt_util.h index da86b081..b2e4e47a 100644 --- a/plugins/mqtt/mqtt_util.h +++ b/plugins/mqtt/mqtt_util.h @@ -27,8 +27,11 @@ extern "C" { #include "connection/mqtt_client_intf.h" #include -int mqtt_option_init(neu_plugin_t *plugin, char *config, - neu_mqtt_option_t *option); +int mqtt_option_init(neu_plugin_t *plugin, char *config, + neu_mqtt_option_t *option); + +const char *mqtt_option_error(int error); + void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option); #ifdef __cplusplus diff --git a/tests/mqtt_test.cc b/tests/mqtt_test.cc index 0f3d8079..23c0aeaa 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -11,155 +11,157 @@ 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 = - // (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); + char *config = (char *) "{\"node_id\":5,\"params\":{}}"; + int rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(3, 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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"\"}}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(3, 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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\"}}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(4, 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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", \"heartbeat-topic\":\"\"}}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(4, 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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\"}}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(5, 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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1}}"; + rc = mqtt_option_init(plugin, config, &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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":false}}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(8, 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); + config = + (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":false, \"host\":\"192.168.10.116\"}}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(9, rc); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":false, " + "\"host\":\"192.168.10.116\", \"port\":1883}}"; + 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 *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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":false, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\"}}"; + 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 *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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":false, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\"}}"; + 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 *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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":true, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\"}}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(6, 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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":true, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\", \"ca\":\"\"}}"; + 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 *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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":true, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\", \"ca\":\"\", " + "\"cert\":\"\" }}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(7, 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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":true, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\", \"ca\":\"\", " + "\"cert\":\"\", \"key\":\"\" }}"; + 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 *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); + config = (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"ssl\":true, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\", \"ca\":\"\", " + "\"cert\":\"\", \"key\":\"\", \"keypass\":\"\" }}"; + rc = mqtt_option_init(plugin, config, &option); EXPECT_EQ(0, rc); mqtt_option_uninit(plugin, &option);