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

Merge pull request #910 from eeff/eeff/dev

Browse Source 
feat(nano-mqtt): handling heartbeat generation
This commit is contained in:
fengzeroz
2022-11-15 11:35:52 +08:00
committed by GitHub
parent 9402cc6d52 a8cfeaa268
commit 362e0a0817
5 changed files with 224 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client.h
+3 -3
View File
@@ -39,9 +39,9 @@ typedef enum {
} neu_mqtt_version_e;
typedef enum {
MQTT_QOS0,
MQTT_QOS1,
MQTT_QOS2,
NEU_MQTT_QOS0,
NEU_MQTT_QOS1,
NEU_MQTT_QOS2,
} neu_mqtt_qos_e;
typedef struct neu_mqtt_client_s neu_mqtt_client_t;
plugins/nano-mqtt/mqtt_handle.c
+66 -2
View File
@@ -17,9 +17,42 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "errcodes.h"
#include "version.h"
#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;
}
int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_error_t *data)
{
@@ -48,7 +81,38 @@ int handle_trans_data(neu_plugin_t * plugin,
int handle_nodes_state(neu_plugin_t *plugin, neu_reqresp_nodes_state_t *states)
{
(void) plugin;
int rv = 0;
char *json_str = NULL;
if (NULL == plugin->client) {
rv = NEU_ERR_MQTT_IS_NULL;
goto end;
}
if (!neu_mqtt_client_is_connected(plugin->client)) {
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;
}
const char * topic = plugin->config.heartbeat_topic;
neu_mqtt_qos_e qos = NEU_MQTT_QOS0;
rv = neu_mqtt_client_publish(plugin->client, qos, topic, json_str,
strlen(json_str));
if (0 != rv) {
plog_error(plugin, "heartbeat pub [%s, QoS%d] error", topic, qos);
rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
}
end:
free(json_str);
utarray_free(states->states);
return 0;
return rv;
}
plugins/nano-mqtt/mqtt_plugin.c
+148 -11
View File
@@ -23,6 +23,21 @@
const neu_plugin_module_t neu_plugin_module;
static void connect_cb(void *data)
{
neu_plugin_t *plugin = data;
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
plog_info(plugin, "plugin `%s` connected", neu_plugin_module.module_name);
}
static void disconnect_cb(void *data)
{
neu_plugin_t *plugin = data;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
plog_info(plugin, "plugin `%s` disconnected",
neu_plugin_module.module_name);
}
static neu_plugin_t *mqtt_plugin_open(void)
{
neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
@@ -49,40 +64,162 @@ static int mqtt_plugin_init(neu_plugin_t *plugin)
static int mqtt_plugin_uninit(neu_plugin_t *plugin)
{
mqtt_config_fini(&plugin->config);
if (plugin->client) {
neu_mqtt_client_close(plugin->client);
neu_mqtt_client_free(plugin->client);
plugin->client = NULL;
}
plog_info(plugin, "uninitialize plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
const mqtt_config_t *config)
{
int rv = 0;
if (NULL == client) {
return 0;
}
rv = neu_mqtt_client_set_zlog_category(client, plugin->common.log);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_zlog_category fail");
return -1;
}
rv = neu_mqtt_client_set_connect_cb(client, connect_cb, plugin);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_connect_cb fail");
return -1;
}
rv = neu_mqtt_client_set_disconnect_cb(client, disconnect_cb, plugin);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_disconnect_cb fail");
return -1;
}
rv = neu_mqtt_client_set_msg_cache_limit(client, config->cache);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_msg_cache_limit fail");
return -1;
}
if (NULL != config->username) {
rv = neu_mqtt_client_set_user(client, config->username,
config->password);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_user fail");
}
}
if (NULL != config->ca) {
rv = neu_mqtt_client_set_tls(client, config->ca, config->cert,
config->key, config->keypass);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_tsl fail");
return -1;
}
}
return rv;
}
static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
{
int rv = 0;
const char *plugin_name = neu_plugin_module.module_name;
int rv = 0;
const char * plugin_name = neu_plugin_module.module_name;
mqtt_config_t config = { 0 };
bool started = false;
rv = mqtt_config_parse(plugin, setting, &config);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_config_parse fail");
return NEU_ERR_NODE_SETTING_INVALID;
}
if (plugin->client && neu_mqtt_client_is_opened(plugin->client)) {
started = true;
rv = neu_mqtt_client_close(plugin->client);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_close fail");
rv = NEU_ERR_EINTERNAL;
goto error;
}
} else {
plugin->client = neu_mqtt_client_new(config.host, config.port,
NEU_MQTT_VERSION_V311);
if (NULL == plugin->client) {
plog_error(plugin, "neu_mqtt_client_new fail");
rv = NEU_ERR_EINTERNAL;
goto error;
}
}
rv = config_mqtt_client(plugin, plugin->client, &config);
if (0 != rv) {
rv = NEU_ERR_MQTT_INIT_FAILURE;
goto error;
}
if (started && 0 != neu_mqtt_client_open(plugin->client)) {
plog_error(plugin, "neu_mqtt_client_open fail");
rv = NEU_ERR_MQTT_CONNECT_FAILURE;
goto error;
}
if (plugin->config.host) {
// already configured
mqtt_config_fini(&plugin->config);
}
rv = mqtt_config_parse(plugin, setting, &plugin->config);
if (0 != rv) {
plog_error(plugin, "config plugin `%s` fail", plugin_name);
return NEU_ERR_NODE_SETTING_INVALID;
}
memmove(&plugin->config, &config, sizeof(config));
plog_info(plugin, "config plugin `%s` success", plugin_name);
return 0;
error:
plog_info(plugin, "config plugin `%s` fail", plugin_name);
mqtt_config_fini(&config);
return rv;
}
static int mqtt_plugin_start(neu_plugin_t *plugin)
{
(void) plugin;
return NEU_ERR_SUCCESS;
int rv = 0;
const char *plugin_name = neu_plugin_module.module_name;
if (NULL == plugin->client) {
plog_error(plugin, "mqtt client is NULL due to init failure");
rv = NEU_ERR_MQTT_IS_NULL;
goto end;
}
if (0 != neu_mqtt_client_open(plugin->client)) {
plog_error(plugin, "neu_mqtt_client_open fail");
rv = NEU_ERR_MQTT_CONNECT_FAILURE;
goto end;
}
end:
if (0 == rv) {
plog_info(plugin, "start plugin `%s` success", plugin_name);
} else {
plog_error(plugin, "start plugin `%s` failed", plugin_name);
}
return rv;
}
static int mqtt_plugin_stop(neu_plugin_t *plugin)
{
(void) plugin;
if (plugin->client) {
neu_mqtt_client_close(plugin->client);
plog_info(plugin, "mqtt client closed");
}
plog_info(plugin, "stop plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
plugins/nano-mqtt/mqtt_plugin.h
+2
View File
@@ -24,6 +24,7 @@
extern "C" {
#endif
#include "connection/mqtt_client.h"
#include "neuron.h"
#include "mqtt_config.h"
@@ -31,6 +32,7 @@ extern "C" {
struct neu_plugin {
neu_plugin_common_t common;
mqtt_config_t config;
neu_mqtt_client_t * client;
};
#ifdef __cplusplus
src/connection/mqtt_client.c
+5
View File
@@ -293,6 +293,10 @@ int neu_mqtt_client_set_msg_cache_limit(neu_mqtt_client_t *client,
{
check_opened();
if (0 == cache_limit) {
return 0;
}
if (NULL == client->sqlite_cfg) {
client->sqlite_cfg = alloc_sqlite_config(client);
if (NULL == client->sqlite_cfg) {
@@ -398,6 +402,7 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
int rv = nng_sendmsg(client->sock, pub_msg, NNG_FLAG_NONBLOCK);
if (0 != rv) {
log(error, "nng_sendmsg fail: %s", nng_strerror(rv));
nng_msg_free(pub_msg);
return -1;
}