GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): add write request/response topic config

Browse Source
This commit is contained in:
eeff
2023-05-05 14:49:20 +08:00
parent 0556e30343
commit 125324546f
3 changed files with 108 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/mqtt.json
+25 -1
View File
@@ -57,6 +57,30 @@
]
}
},
"write-req-topic": {
"name": "Write Request Topic",
"name_zh": "写请求主题",
"description": "MQTT topic to receive write request messages.",
"description_zh": "接收点位写入请求的 MQTT 主题。",
"attribute": "required",
"type": "string",
"default": "/neuron/${node-name}/write/req",
"valid": {
"length": 255
}
},
"write-resp-topic": {
"name": "Write Response Topic",
"name_zh": "写响应主题",
"description": "MQTT topic to send write response messages.",
"description_zh": "发送点位写入响应的 MQTT 主题。",
"attribute": "required",
"type": "string",
"default": "/neuron/${node-name}/write/resp",
"valid": {
"length": 255
}
},
"offline-cache": {
"name": "Offline Data Caching",
"name_zh": "离线缓存",
@@ -218,4 +242,4 @@
"length": 256
}
}
}
}
plugins/mqtt/mqtt_config.c
+67 -30
View File
@@ -17,6 +17,7 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "utils/asprintf.h"
#include "json/json.h"
#include "json/neu_json_param.h"
@@ -203,6 +204,18 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT };
neu_json_elem_t write_req_topic = {
.name = "write-req-topic",
.t = NEU_JSON_STR,
.v.val_str = NULL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t write_resp_topic = {
.name = "write-resp-topic",
.t = NEU_JSON_STR,
.v.val_str = NULL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t offline_cache = { .name = "offline-cache",
.t = NEU_JSON_BOOL };
neu_json_elem_t cache_mem_size = { .name = "cache-mem-size",
@@ -224,8 +237,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
return -1;
}
ret = neu_parse_param(setting, &err_param, 5, &client_id, &format, &qos,
&host, &port);
ret = neu_parse_param(setting, &err_param, 7, &client_id, &qos, &format,
&write_req_topic, &write_resp_topic, &host, &port);
if (0 != ret) {
plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
goto error;
@@ -251,6 +264,22 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
// write request topic
if (NULL == write_req_topic.v.val_str &&
0 > neu_asprintf(&write_req_topic.v.val_str, "/neuron/%s/write/req",
plugin->common.name)) {
plog_error(plugin, "setting write request topic error");
goto error;
}
// write response topic
if (NULL == write_resp_topic.v.val_str &&
0 > neu_asprintf(&write_resp_topic.v.val_str, "/neuron/%s/write/resp",
plugin->common.name)) {
plog_error(plugin, "setting write request topic error");
goto error;
}
// offline cache
ret = parse_cache_params(plugin, setting, &offline_cache, &cache_mem_size,
&cache_disk_size);
@@ -287,55 +316,61 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->cache = offline_cache.v.val_bool;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
config->cache_disk_size = cache_disk_size.v.val_int * MB;
config->host = host.v.val_str;
config->port = port.v.val_int;
config->username = username.v.val_str;
config->password = password.v.val_str;
config->ca = ca.v.val_str;
config->cert = cert.v.val_str;
config->key = key.v.val_str;
config->keypass = keypass.v.val_str;
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->write_req_topic = write_req_topic.v.val_str;
config->write_resp_topic = write_resp_topic.v.val_str;
config->cache = offline_cache.v.val_bool;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
config->cache_disk_size = cache_disk_size.v.val_int * MB;
config->host = host.v.val_str;
config->port = port.v.val_int;
config->username = username.v.val_str;
config->password = password.v.val_str;
config->ca = ca.v.val_str;
config->cert = cert.v.val_str;
config->key = key.v.val_str;
config->keypass = keypass.v.val_str;
plog_info(plugin, "config client-id : %s", config->client_id);
plog_info(plugin, "config qos : %d", config->qos);
plog_info(plugin, "config format : %s",
plog_info(plugin, "config client-id : %s", config->client_id);
plog_info(plugin, "config qos : %d", config->qos);
plog_info(plugin, "config format : %s",
mqtt_upload_format_str(config->format));
plog_info(plugin, "config cache : %zu", config->cache);
plog_info(plugin, "config cache-mem-size : %zu", config->cache_mem_size);
plog_info(plugin, "config cache-disk-size: %zu", config->cache_disk_size);
plog_info(plugin, "config host : %s", config->host);
plog_info(plugin, "config port : %" PRIu16, config->port);
plog_info(plugin, "config write-req-topic : %s", config->write_req_topic);
plog_info(plugin, "config write-resp-topic: %s", config->write_req_topic);
plog_info(plugin, "config cache : %zu", config->cache);
plog_info(plugin, "config cache-mem-size : %zu", config->cache_mem_size);
plog_info(plugin, "config cache-disk-size : %zu", config->cache_disk_size);
plog_info(plugin, "config host : %s", config->host);
plog_info(plugin, "config port : %" PRIu16, config->port);
if (config->username) {
plog_info(plugin, "config username : %s", config->username);
plog_info(plugin, "config username : %s", config->username);
}
if (config->password) {
plog_info(plugin, "config password : %s",
plog_info(plugin, "config password : %s",
0 == strlen(config->password) ? "" : placeholder);
}
if (config->ca) {
plog_info(plugin, "config ca : %s", placeholder);
plog_info(plugin, "config ca : %s", placeholder);
}
if (config->cert) {
plog_info(plugin, "config cert : %s", placeholder);
plog_info(plugin, "config cert : %s", placeholder);
}
if (config->key) {
plog_info(plugin, "config key : %s", placeholder);
plog_info(plugin, "config key : %s", placeholder);
}
if (config->keypass) {
plog_info(plugin, "config keypass : %s", placeholder);
plog_info(plugin, "config keypass : %s", placeholder);
}
return 0;
error:
free(err_param);
free(write_req_topic.v.val_str);
free(write_resp_topic.v.val_str);
free(host.v.val_str);
free(username.v.val_str);
free(password.v.val_str);
@@ -349,6 +384,8 @@ error:
void mqtt_config_fini(mqtt_config_t *config)
{
free(config->client_id);
free(config->write_req_topic);
free(config->write_resp_topic);
free(config->host);
free(config->username);
free(config->password);
plugins/mqtt/mqtt_config.h
+16 -14
View File
@@ -48,20 +48,22 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
}
typedef struct {
char * client_id; // client id
mqtt_upload_format_e format; // upload format
neu_mqtt_qos_e qos; // message QoS
size_t cache; // cache enable flag
size_t cache_mem_size; // cache memory size in bytes
size_t cache_disk_size; // cache disk size in bytes
char * host; // broker host
uint16_t port; // broker port
char * username; // user name
char * password; // user password
char * ca; // CA
char * cert; // client cert
char * key; // client key
char * keypass; // client key password
char * client_id; // client id
neu_mqtt_qos_e qos; // message QoS
mqtt_upload_format_e format; // upload format
char * write_req_topic; // write request topic
char * write_resp_topic; // write response topic
size_t cache; // cache enable flag
size_t cache_mem_size; // cache memory size in bytes
size_t cache_disk_size; // cache disk size in bytes
char * host; // broker host
uint16_t port; // broker port
char * username; // user name
char * password; // user password
char * ca; // CA
char * cert; // client cert
char * key; // client key
char * keypass; // client key password
} mqtt_config_t;
int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,