GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-22 01:05:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

mqtt: config enable_topic

Browse Source
This commit is contained in:
fengzero
2025-06-12 02:51:42 +00:00
parent f0b746a1ab
commit 8ee7ab8a36
4 changed files with 46 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/mqtt.json
+10
View File
@@ -109,6 +109,16 @@
"default": true,
"valid": {}
},
"enable_topic": {
"name": "Enable driver topic",
"name_zh": "启动驱动相关主题",
"description": "Subscription driver related topics, including read/req, write/req and action/req.",
"description_zh": "是否订阅驱动相关主题，包括 read/req、write/req 和 action/req。",
"attribute": "optional",
"type": "bool",
"default": true,
"valid": {}
},
"write-req-topic": {
"name": "Write Request Topic",
"name_zh": "写请求主题",
plugins/mqtt/mqtt_config.c
+19 -7
View File
@@ -291,13 +291,19 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
.t = NEU_JSON_INT };
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 ssl = { .name = "ssl", .t = NEU_JSON_BOOL };
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 };
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 ssl = { .name = "ssl", .t = NEU_JSON_BOOL };
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 };
neu_json_elem_t enable_topic = {
.name = "enable_topic",
.t = NEU_JSON_BOOL,
.v.val_bool = true,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
};
neu_json_elem_t upload_drv_state = {
.name = "upload_drv_state",
.t = NEU_JSON_BOOL,
@@ -374,6 +380,11 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
free(schema.v.val_str);
}
ret = neu_parse_param(setting, &err_param, 1, &enable_topic);
if (0 != ret) {
enable_topic.v.val_bool = true; // default to true
}
ret = neu_parse_param(setting, &err_param, 1, &driver_topic_prefix);
if (0 != ret) {
plog_warn(plugin, "parsing driver topic prefix fail, key: `%s`",
@@ -456,6 +467,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->enable_topic = enable_topic.v.val_bool;
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;
plugins/mqtt/mqtt_config.h
+7 -7
View File
@@ -94,13 +94,14 @@ typedef struct {
} mqtt_driver_topic_t;
typedef struct {
neu_mqtt_version_e version; // mqtt version
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
neu_mqtt_version_e version; // mqtt version
char * client_id; // client id
neu_mqtt_qos_e qos; // message QoS
mqtt_upload_format_e format; // upload format
bool enable_topic; // default true
char * write_req_topic; // write request topic
char * write_resp_topic; // write response topic
char * driver_topic_prefix; // driver topic prefix
mqtt_driver_topic_t driver_topic;
@@ -123,7 +124,6 @@ typedef struct {
char * keypass; // client key password
// remove in 2.6, keep it here
// for backward compatibility
size_t n_schema_vt;
mqtt_schema_vt_t *schema_vts;
} mqtt_config_t;
plugins/mqtt/mqtt_plugin_intf.c
+10 -4
View File
@@ -404,7 +404,9 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
}
} else if (neu_mqtt_client_is_open(plugin->client)) {
started = true;
plugin->unsubscribe(plugin, &plugin->config);
if (plugin->config.enable_topic) {
plugin->unsubscribe(plugin, &plugin->config);
}
rv = neu_mqtt_client_close(plugin->client);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_close fail");
@@ -440,8 +442,10 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
rv = NEU_ERR_EINTERNAL;
goto error;
}
if (0 != (rv = plugin->subscribe(plugin, &config))) {
goto error;
if (plugin->config.enable_topic) {
if (0 != (rv = plugin->subscribe(plugin, &config))) {
goto error;
}
}
}
@@ -499,7 +503,9 @@ end:
int mqtt_plugin_stop(neu_plugin_t *plugin)
{
if (plugin->client) {
plugin->unsubscribe(plugin, &plugin->config);
if (plugin->config.enable_topic) {
plugin->unsubscribe(plugin, &plugin->config);
}
neu_mqtt_client_close(plugin->client);
plog_notice(plugin, "mqtt client closed");
}