diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index a4512ccb..5d007df0 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -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.", diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index 83549046..e6f5b9f9 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/plugins/mqtt/mqtt_config.c @@ -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); diff --git a/plugins/mqtt/mqtt_config.h b/plugins/mqtt/mqtt_config.h index f6bede50..2efa07bd 100644 --- a/plugins/mqtt/mqtt_config.h +++ b/plugins/mqtt/mqtt_config.h @@ -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 diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index f5bab788..f58cb5e5 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -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; -} diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index 319040cc..81d3a58e 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/plugins/mqtt/mqtt_handle.h @@ -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 diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index aa349bc6..f19a0742 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -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, };