GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 07:06:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): remove setting/publishing heartbeat-topic

Browse Source
This commit is contained in:
eeff
2022-12-14 16:42:07 +08:00
parent b3eb9d55bd
commit c08a895b0a
6 changed files with 11 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/mqtt.json
-10
View File
@@ -19,16 +19,6 @@
"length": 255
}
},
"heartbeat-topic": {
"name": "heartbeat topic",
"description": "Topic to which publish heartbeat messages.",
"type": "string",
"attribute": "required",
"default": "/neuron/${node-name}/heartbeat",
"valid": {
"length": 255
}
},
"format": {
"name": "upload format",
"description": "JSON format of the data reported in the upload topic. In values mode, data are split into `values` and `errors` sub objects. In tags mode, tag data are put in a single array.",
plugins/mqtt/mqtt_config.c
+10 -22
View File
@@ -122,13 +122,11 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
char * err_param = NULL;
const char *placeholder = "********";
neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR };
neu_json_elem_t upload_topic = { .name = "upload-topic",
neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR };
neu_json_elem_t upload_topic = { .name = "upload-topic",
.t = NEU_JSON_STR };
neu_json_elem_t heartbeat_topic = { .name = "heartbeat-topic",
.t = NEU_JSON_STR };
neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT };
neu_json_elem_t cache_mem_size = { .name = "cache-mem-size",
neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT };
neu_json_elem_t cache_mem_size = { .name = "cache-mem-size",
.t = NEU_JSON_INT };
neu_json_elem_t cache_disk_size = { .name = "cache-disk-size",
.t = NEU_JSON_INT };
@@ -147,9 +145,9 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
return -1;
}
ret = neu_parse_param(setting, &err_param, 8, &client_id, &upload_topic,
&heartbeat_topic, &format, &cache_mem_size,
&cache_disk_size, &host, &port);
ret = neu_parse_param(setting, &err_param, 7, &client_id, &upload_topic,
&format, &cache_mem_size, &cache_disk_size, &host,
&port);
if (0 != ret) {
plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
goto error;
@@ -167,12 +165,6 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
// heartbeat-topic, required
if (0 == strlen(heartbeat_topic.v.val_str)) {
plog_error(plugin, "setting empty heartbeat-topic");
goto error;
}
// format, required
if (MQTT_UPLOAD_FORMAT_VALUES != format.v.val_int &&
MQTT_UPLOAD_FORMAT_TAGS != format.v.val_int) {
@@ -239,10 +231,9 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
config->client_id = client_id.v.val_str;
config->upload_topic = upload_topic.v.val_str;
config->heartbeat_topic = heartbeat_topic.v.val_str;
config->format = format.v.val_int;
config->client_id = client_id.v.val_str;
config->upload_topic = upload_topic.v.val_str;
config->format = format.v.val_int;
config->cache =
0 != cache_mem_size.v.val_int || 0 != cache_disk_size.v.val_int;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
@@ -258,7 +249,6 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
plog_info(plugin, "config client-id : %s", config->client_id);
plog_info(plugin, "config upload-topic : %s", config->upload_topic);
plog_info(plugin, "config heartbeat-topic: %s", config->heartbeat_topic);
plog_info(plugin, "config format : %s",
mqtt_upload_format_str(config->format));
plog_info(plugin, "config cache : %zu", config->cache);
@@ -292,7 +282,6 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
error:
free(err_param);
free(upload_topic.v.val_str);
free(heartbeat_topic.v.val_str);
free(host.v.val_str);
free(username.v.val_str);
free(password.v.val_str);
@@ -307,7 +296,6 @@ void mqtt_config_fini(mqtt_config_t *config)
{
free(config->client_id);
free(config->upload_topic);
free(config->heartbeat_topic);
free(config->host);
free(config->username);
free(config->password);
plugins/mqtt/mqtt_config.h
-1
View File
@@ -49,7 +49,6 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
typedef struct {
char * client_id; // client id
char * upload_topic; // upload topic
char * heartbeat_topic; // heartbeat topic
mqtt_upload_format_e format; // upload format
size_t cache; // cache enable flag
size_t cache_mem_size; // cache memory size in bytes
plugins/mqtt/mqtt_handle.c
-65
View File
@@ -26,36 +26,6 @@
#include "mqtt_handle.h"
#include "mqtt_plugin.h"
static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states)
{
(void) plugin;
char * version = NEURON_VERSION;
neu_json_states_head_t header = { .version = version,
.timpstamp = global_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));
if (NULL == json.states) {
return NULL;
}
utarray_foreach(states, neu_nodes_state_t *, state)
{
int index = utarray_eltidx(states, state);
json.states[index].node = state->node;
json.states[index].link = state->state.link;
json.states[index].running = state->state.running;
}
neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header,
neu_json_encode_state_header_resp, &json_str);
free(json.states);
return json_str;
}
static int tag_values_to_json(neu_resp_tag_value_t *tags, uint16_t len,
neu_json_read_resp_t *json)
{
@@ -524,38 +494,3 @@ int handle_trans_data(neu_plugin_t * plugin,
return rv;
}
int handle_nodes_state(neu_plugin_t *plugin, neu_reqresp_nodes_state_t *states)
{
int rv = 0;
char *json_str = NULL;
if (NULL == plugin->client) {
rv = NEU_ERR_MQTT_IS_NULL;
goto end;
}
if (0 == plugin->config.cache &&
!neu_mqtt_client_is_connected(plugin->client)) {
// cache disable and we are disconnected
rv = NEU_ERR_MQTT_FAILURE;
goto end;
}
json_str = generate_heartbeat_json(plugin, states->states);
if (NULL == json_str) {
plog_error(plugin, "generate heartbeat json fail");
rv = NEU_ERR_EINTERNAL;
goto end;
}
char * topic = plugin->config.heartbeat_topic;
neu_mqtt_qos_e qos = NEU_MQTT_QOS0;
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
json_str = NULL;
end:
utarray_free(states->states);
return rv;
}
plugins/mqtt/mqtt_handle.h
-2
View File
@@ -42,8 +42,6 @@ int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
int handle_trans_data(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *trans_data);
int handle_nodes_state(neu_plugin_t *plugin, neu_reqresp_nodes_state_t *states);
#ifdef __cplusplus
}
#endif
plugins/mqtt/mqtt_plugin.c
+1 -5
View File
@@ -349,10 +349,6 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
case NEU_REQRESP_TRANS_DATA:
error = handle_trans_data(plugin, data);
break;
case NEU_REQRESP_NODES_STATE: {
error = handle_nodes_state(plugin, data);
break;
}
case NEU_REQRESP_NODE_DELETED:
break;
default:
@@ -385,7 +381,7 @@ const neu_plugin_module_t neu_plugin_module = {
.intf_funs = &plugin_intf_funs,
.kind = NEU_PLUGIN_KIND_SYSTEM,
.type = NEU_NA_TYPE_APP,
.sub_msg[0] = NEU_SUBSCRIBE_NODES_STATE,
.sub_msg = { 0 },
.display = true,
.single = false,
};