GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-26 11:56:10 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
v2.2
neuron/plugins/mqtt/mqtt_plugin.c
T

684 lines
20 KiB
C
Raw Permalink Blame History

/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <assert.h>
#include "command/command.h"
#include "mqtt_plugin.h"
#include "mqtt_util.h"
#define INTERVAL 100000U
#define TIMEOUT 3000U
#define MAX_TOPIC_LEN 256
#define TOPIC_READ_REQ "/neuron/%s/read/req"
#define TOPIC_WRITE_REQ "/neuron/%s/write/req"
#define TOPIC_READ_RES "/neuron/%s/read/resp"
#define TOPIC_WRITE_RES "/neuron/%s/write/resp"
#define TOPIC_UPLOAD_RES "/neuron/%s/upload"
#define TOPIC_HEARTBEAT_RES "/neuron/%s/heartbeat"
#define QOS0 0
#define QOS1 1
#define QOS2 2
const neu_plugin_module_t neu_plugin_module;
struct topic_pair {
char *topic_request;
char *topic_response;
int qos_request;
int qos_response;
int type;
};
struct mqtt_routine {
neu_plugin_t * plugin;
neu_mqtt_option_t option;
neu_mqtt_client_t client;
UT_array * topics;
};
static char *topics_format(char *format, char *name)
{
char *topic = calloc(1, MAX_TOPIC_LEN);
if (NULL == topic) {
return NULL;
}
int rc = snprintf(topic, MAX_TOPIC_LEN, format, name);
if (-1 != rc) {
return topic;
}
free(topic);
return NULL;
}
static void topics_add(UT_array *topics, neu_plugin_t *plugin, char *request,
int qos_request, char *response, int qos_response,
int type)
{
struct topic_pair *pair = calloc(1, sizeof(struct topic_pair));
pair->topic_request = request;
pair->topic_response = response;
pair->qos_request = qos_request;
pair->qos_response = qos_response;
pair->type = type;
utarray_push_back(topics, &pair);
plog_info(plugin, "add topic-req:%s, topic-res:%s, count:%d",
pair->topic_request, pair->topic_response, utarray_len(topics));
}
static void topics_cleanup(UT_array *topics, neu_plugin_t *plugin)
{
for (struct topic_pair **pair =
(struct topic_pair **) utarray_front(topics);
NULL != pair;
pair = (struct topic_pair **) utarray_next(topics, pair)) {
plog_info(plugin, "remove topic-req:%s, topic-res:%s",
(*pair)->topic_request, (*pair)->topic_response);
if (NULL != (*pair)->topic_request) {
free((*pair)->topic_request);
}
if (NULL != (*pair)->topic_response) {
free((*pair)->topic_response);
}
free(*pair);
}
}
static void topics_generate(UT_array *topics, neu_plugin_t *plugin,
neu_mqtt_option_t *option)
{
char *name = plugin->common.name;
// READ TOPIC
char *read_req = topics_format(TOPIC_READ_REQ, name);
char *read_res = topics_format(TOPIC_READ_RES, name);
topics_add(topics, plugin, read_req, QOS0, read_res, QOS0, TOPIC_TYPE_READ);
// WRITE TOPIC
char *write_req = topics_format(TOPIC_WRITE_REQ, name);
char *write_res = topics_format(TOPIC_WRITE_RES, name);
topics_add(topics, plugin, write_req, QOS0, write_res, QOS0,
TOPIC_TYPE_WRITE);
/// UPLOAD TOPIC SETTING
char *upload_res = strdup(option->upload_topic);
topics_add(topics, plugin, NULL, QOS0, upload_res, QOS0, TOPIC_TYPE_UPLOAD);
// HEARTBEAT TOPIC SETTING
char *heartbeat_res = strdup(option->heartbeat_topic);
topics_add(topics, plugin, NULL, QOS0, heartbeat_res, QOS0,
TOPIC_TYPE_HEARTBEAT);
}
static struct topic_pair *topics_find_topic(UT_array *topics, const char *topic)
{
for (struct topic_pair **pair =
(struct topic_pair **) utarray_front(topics);
NULL != pair;
pair = (struct topic_pair **) utarray_next(topics, pair)) {
if (0 == strcmp((*pair)->topic_request, topic)) {
return *pair;
}
}
return NULL;
}
static struct topic_pair *topics_find_type(UT_array *topics, int type)
{
for (struct topic_pair **pair =
(struct topic_pair **) utarray_front(topics);
NULL != pair;
pair = (struct topic_pair **) utarray_next(topics, pair)) {
if (type == (*pair)->type) {
return *pair;
}
}
return NULL;
}
static void mqtt_routine_response(const char *topic_name, size_t topic_len,
void *payload, const size_t len,
void *context)
{
neu_plugin_t * plugin = (neu_plugin_t *) context;
mqtt_routine_t * routine = plugin->routine;
struct topic_pair *pair = topics_find_topic(routine->topics, topic_name);
mqtt_response_t response = { .topic_name = topic_name,
.topic_len = topic_len,
.payload = payload,
.len = len,
.plugin = context,
.topic_pair = pair,
.type = pair->type };
command_response_handle(&response);
}
static void topics_subscribe(UT_array *topics, neu_mqtt_client_t client)
{
for (struct topic_pair **pair =
(struct topic_pair **) utarray_front(topics);
NULL != pair;
pair = (struct topic_pair **) utarray_next(topics, pair)) {
if (NULL != (*pair)->topic_request) {
neu_mqtt_client_subscribe(client, (*pair)->topic_request,
(*pair)->qos_request,
mqtt_routine_response);
}
}
}
static void mqtt_routine_state(void *context, int state)
{
if (NULL == context) {
return;
}
neu_plugin_t *plugin = (neu_plugin_t *) context;
if (0 == state) {
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
} else {
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
}
}
static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin,
const char * config)
{
assert(NULL != plugin);
assert(NULL != config);
mqtt_routine_t *routine = calloc(1, sizeof(mqtt_routine_t));
if (NULL == routine) {
return NULL;
}
// MQTT client start
int rc = mqtt_option_init(plugin, (char *) config, &routine->option);
if (0 != rc) {
plog_error(plugin, "%s", mqtt_option_error(rc));
mqtt_option_uninit(plugin, &routine->option);
free(routine);
return NULL;
}
routine->option.state_update_func = mqtt_routine_state;
routine->option.log = plugin->common.log;
neu_mqtt_client_t *client_p = (neu_mqtt_client_t *) &routine->client;
neu_err_code_e error = NEU_ERR_SUCCESS;
error = neu_mqtt_client_open(client_p, &routine->option, plugin);
if (NEU_ERR_SUCCESS != error) {
mqtt_option_uninit(plugin, &routine->option);
free(routine);
return NULL;
}
neu_mqtt_client_continue(routine->client);
plog_info(plugin, "open mqtt client: %s:%s, code:%d", routine->option.host,
routine->option.port, error);
// topics generate
routine->plugin = plugin;
UT_icd topic_ptr_icd = { sizeof(struct topic_pair *), NULL, NULL, NULL };
utarray_new(routine->topics, &topic_ptr_icd);
topics_generate(routine->topics, plugin, &routine->option);
topics_subscribe(routine->topics, routine->client);
return routine;
}
static void mqtt_routine_stop(mqtt_routine_t *routine)
{
assert(NULL != routine);
// stop recevied
neu_mqtt_client_suspend(routine->client);
// close mqtt client and clean
neu_mqtt_client_close(routine->client);
plog_info(routine->plugin, "close mqtt client:%s:%s", routine->option.host,
routine->option.port);
topics_cleanup(routine->topics, routine->plugin);
utarray_free(routine->topics);
mqtt_option_uninit(routine->plugin, &routine->option);
}
static neu_plugin_t *mqtt_plugin_open(void)
{
neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
neu_plugin_common_init(&plugin->common);
return plugin;
}
static int mqtt_plugin_close(neu_plugin_t *plugin)
{
assert(NULL != plugin);
const char *name = neu_plugin_module.module_name;
plog_info(plugin, "success to free plugin:%s", name);
free(plugin);
return NEU_ERR_SUCCESS;
}
static void plugin_cache_init(neu_plugin_t *plugin);
static int mqtt_plugin_init(neu_plugin_t *plugin)
{
assert(NULL != plugin);
plugin->routine = NULL;
plugin->running = false;
plugin->config = NULL;
plugin_cache_init(plugin);
const char *name = neu_plugin_module.module_name;
plog_info(plugin, "initialize plugin: %s", name);
return NEU_ERR_SUCCESS;
}
static int plguin_start_running(neu_plugin_t *plugin)
{
if (!plugin->running && NULL != plugin->config) {
mqtt_routine_t *routine = mqtt_routine_start(plugin, plugin->config);
if (NULL == routine) {
plugin->routine = NULL;
plugin->running = false;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
return -1;
}
plugin->routine = routine;
plugin->running = true;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
}
return 0;
}
static void plugin_stop_running(neu_plugin_t *plugin)
{
if (plugin->running) {
mqtt_routine_stop(plugin->routine);
if (NULL != plugin->routine) {
free(plugin->routine);
plugin->routine = NULL;
}
plugin->running = false;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
}
}
static void plugin_config_free(neu_plugin_t *plugin)
{
if (NULL != plugin->config) {
free(plugin->config);
plugin->config = NULL;
}
}
static void plugin_config_save(neu_plugin_t *plugin, const char *config)
{
plugin_config_free(plugin);
plugin->config = strdup(config);
}
static void plugin_cache_init(neu_plugin_t *plugin)
{
plugin->cache = neu_mem_cache_create(0, 0);
nng_mtx_alloc(&plugin->mutex);
}
static void plugin_cache_uninit(neu_plugin_t *plugin)
{
nng_mtx_free(plugin->mutex);
neu_mem_cache_destroy(plugin->cache);
}
static int mqtt_plugin_uninit(neu_plugin_t *plugin)
{
assert(NULL != plugin);
plugin_stop_running(plugin);
plugin_cache_uninit(plugin);
plugin_config_free(plugin);
const char *name = neu_plugin_module.module_name;
plog_info(plugin, "uninitialize plugin: %s", name);
return NEU_ERR_SUCCESS;
}
static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config)
{
assert(NULL != plugin);
assert(NULL != config);
// Option setting
if (NEU_ERR_SUCCESS != mqtt_option_validate(plugin, config)) {
return NEU_ERR_NODE_SETTING_INVALID;
}
plugin_config_save(plugin, config);
size_t cache_size = mqtt_option_item_cache(plugin, config);
size_t max_bytes = cache_size * 1024 * 1024;
nng_mtx_lock(plugin->mutex);
neu_mem_cache_resize(plugin->cache, max_bytes, 0);
nng_mtx_unlock(plugin->mutex);
// Routine not running
if (!plugin->running) {
return NEU_ERR_SUCCESS;
}
plugin_stop_running(plugin);
int rc = plguin_start_running(plugin);
if (0 != rc) {
return NEU_ERR_MQTT_INIT_FAILURE;
}
plog_info(plugin, "config plugin: %s", neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
}
static int mqtt_plugin_start(neu_plugin_t *plugin)
{
assert(NULL != plugin);
int rc = plguin_start_running(plugin);
if (0 != rc) {
return NEU_ERR_NODE_SETTING_INVALID;
}
return NEU_ERR_SUCCESS;
}
static int mqtt_plugin_stop(neu_plugin_t *plugin)
{
assert(NULL != plugin);
plugin_stop_running(plugin);
return NEU_ERR_SUCCESS;
}
static neu_err_code_e write_response(neu_plugin_t * plugin,
neu_reqresp_head_t *head,
neu_resp_error_t * data)
{
neu_resp_error_t *write_data = data;
if (!plugin->running) {
return NEU_ERR_MQTT_FAILURE;
}
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_WRITE;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = command_write_response(plugin, head, write_data);
const char *topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(plugin, "write response publish error code :%d, topoic:%s",
error, topic);
free(json_str);
return error;
}
plog_info(plugin, "topic:%s, qos:%d, payload:%s", topic, qos, json_str);
free(json_str);
return NEU_ERR_SUCCESS;
}
static neu_err_code_e read_response(neu_plugin_t * plugin,
neu_reqresp_head_t *head, void *data)
{
neu_resp_read_group_t *read_data = data;
if (!plugin->running) {
return NEU_ERR_MQTT_FAILURE;
}
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = command_read_once_response(plugin, head, read_data,
routine->option.format);
const char *topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(plugin, "read response publish error code :%d, topoic:%s",
error, topic);
free(json_str);
return error;
}
plog_info(plugin, "topic:%s, qos:%d, payload:%s", topic, qos, json_str);
free(json_str);
return NEU_ERR_SUCCESS;
}
static void cache_string_release(void *data)
{
free(data);
}
static int cache_send(neu_plugin_t *plugin)
{
mqtt_routine_t *routine = plugin->routine;
nng_mtx_lock(plugin->mutex);
cache_item_t item = neu_mem_cache_earliest(plugin->cache);
if (NULL == item.data) {
nng_mtx_unlock(plugin->mutex);
return -1;
}
int type = TOPIC_TYPE_UPLOAD;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = item.data;
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(plugin, "cache publish error code :%d, topoic:%s", error,
topic);
}
free(json_str);
nng_mtx_unlock(plugin->mutex);
return 0;
}
static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
{
neu_reqresp_trans_data_t *trans_data = data;
if (!plugin->running) {
return NEU_ERR_MQTT_FAILURE;
}
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_UPLOAD;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char *json_str = command_read_periodic_response(plugin, trans_data,
routine->option.format);
if (NULL == json_str) {
return NEU_ERR_MQTT_FAILURE;
}
neu_err_code_e rc = neu_mqtt_client_is_connected(routine->client);
if (0 != rc) {
cache_item_t item = {
.size = strlen(json_str) + 1,
.data = json_str,
.release = cache_string_release,
};
nng_mtx_lock(plugin->mutex);
rc = neu_mem_cache_add(plugin->cache, &item);
nng_mtx_unlock(plugin->mutex);
if (0 != rc) {
item.release(item.data);
}
return NEU_ERR_MQTT_CONNECT_FAILURE;
} else {
for (int i = 0; i < 100; i++) {
int r = cache_send(plugin);
if (-1 == r) {
break;
}
}
}
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(plugin, "trans data publish error code :%d, topoic:%s",
error, topic);
free(json_str);
return error;
}
free(json_str);
return NEU_ERR_SUCCESS;
}
static neu_err_code_e node_state_send(neu_plugin_t * plugin,
neu_reqresp_head_t *head, void *data)
{
UNUSED(head);
neu_reqresp_nodes_state_t *states = (neu_reqresp_nodes_state_t *) data;
mqtt_routine_t * routine = plugin->routine;
if (!plugin->running) {
utarray_free(states->states);
return NEU_ERR_MQTT_FAILURE;
}
int type = TOPIC_TYPE_HEARTBEAT;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str =
command_heartbeat_response(routine->plugin, states->states);
if (NULL == json_str) {
utarray_free(states->states);
return -1;
}
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(routine->plugin,
"heartbeat publish error code :%d, topoic:%s", error, topic);
}
free(json_str);
utarray_free(states->states);
return NEU_ERR_SUCCESS;
}
static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data)
{
assert(NULL != plugin);
assert(NULL != head);
assert(NULL != data);
neu_err_code_e error = NEU_ERR_SUCCESS;
switch (head->type) {
case NEU_RESP_ERROR:
error = write_response(plugin, head, data);
break;
case NEU_RESP_READ_GROUP:
error = read_response(plugin, head, data);
break;
case NEU_REQRESP_TRANS_DATA:
error = trans_data(plugin, data);
break;
case NEU_REQRESP_NODES_STATE: {
error = node_state_send(plugin, head, data);
break;
}
case NEU_REQRESP_NODE_DELETED:
break;
default:
error = NEU_ERR_MQTT_FAILURE;
break;
}
return error;
}
static const neu_plugin_intf_funs_t plugin_intf_funs = {
.open = mqtt_plugin_open,
.close = mqtt_plugin_close,
.init = mqtt_plugin_init,
.uninit = mqtt_plugin_uninit,
.start = mqtt_plugin_start,
.stop = mqtt_plugin_stop,
.setting = mqtt_plugin_config,
.request = mqtt_plugin_request,
};
const neu_plugin_module_t neu_plugin_module = {
.version = NEURON_PLUGIN_VER_1_0,
.module_name = "mqtt",
.module_descr =
"Neuron northbound MQTT communication application. The data collected "
"by Neuron from the device can be transmitted to the MQTT Broker "
"through the MQTT application, and users can also send commands to "
"Neuron through the MQTT application.",
.intf_funs = &plugin_intf_funs,
.kind = NEU_PLUGIN_KIND_SYSTEM,
.type = NEU_NA_TYPE_APP,
.sub_msg[0] = NEU_SUBSCRIBE_NODES_STATE,
.single = false,
};
Reference in New Issue View Git Blame Copy Permalink