mirror of
https://github.com/emqx/neuron.git
synced 2026-10-18 13:41:22 +08:00
hxy7yx
395 lines
13 KiB
C
395 lines
13 KiB
C
/**
|
|
* NEURON IIoT System for Industry 4.0
|
|
* Copyright (C) 2020-2024 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 "utils/asprintf.h"
|
|
#include "utils/time.h"
|
|
|
|
#include "mqtt_config.h"
|
|
#include "mqtt_handle.h"
|
|
#include "mqtt_plugin.h"
|
|
#include "mqtt_plugin_intf.h"
|
|
|
|
#define AUTH_SAS 0
|
|
#define AUTH_CERT 1
|
|
|
|
static int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
|
|
neu_json_elem_t *cert, neu_json_elem_t *key)
|
|
{
|
|
// cert, required
|
|
int ret = neu_parse_param(setting, NULL, 1, cert);
|
|
if (0 != ret) {
|
|
plog_notice(plugin, "setting no cert");
|
|
return -1;
|
|
}
|
|
|
|
if (0 != decode_b64_param(plugin, cert)) {
|
|
return -1;
|
|
}
|
|
|
|
// key, required
|
|
ret = parse_b64_param(plugin, setting, key);
|
|
if (0 != ret) {
|
|
return -1;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int azure_parse_config(neu_plugin_t *plugin, const char *setting,
|
|
mqtt_config_t *config)
|
|
{
|
|
int ret = 0;
|
|
char * err_param = NULL;
|
|
const char *placeholder = "********";
|
|
char * write_req_topic = NULL;
|
|
char * write_resp_topic = NULL;
|
|
char * username = NULL;
|
|
|
|
neu_json_elem_t client_id = { .name = "device-id", .t = NEU_JSON_STR };
|
|
neu_json_elem_t qos = { .name = "qos", .t = NEU_JSON_INT };
|
|
neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT };
|
|
neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR };
|
|
neu_json_elem_t auth = { .name = "auth", .t = NEU_JSON_INT };
|
|
neu_json_elem_t password = { .name = "sas", .t = NEU_JSON_STR };
|
|
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 };
|
|
|
|
if (NULL == setting || NULL == config) {
|
|
plog_error(plugin, "invalid argument, null pointer");
|
|
return -1;
|
|
}
|
|
|
|
ret = neu_parse_param(setting, &err_param, 5, &client_id, &qos, &format,
|
|
&host, &auth);
|
|
if (0 != ret) {
|
|
plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
|
|
goto error;
|
|
}
|
|
|
|
// client-id, required
|
|
if (0 == strlen(client_id.v.val_str)) {
|
|
plog_error(plugin, "setting empty client-id");
|
|
goto error;
|
|
}
|
|
|
|
// qos, required
|
|
if (qos.v.val_int != NEU_MQTT_QOS0 && NEU_MQTT_QOS1 != qos.v.val_int) {
|
|
plog_error(plugin, "setting invalid qos: %" PRIi64, qos.v.val_int);
|
|
goto error;
|
|
}
|
|
|
|
// format, required
|
|
if (MQTT_UPLOAD_FORMAT_VALUES != format.v.val_int &&
|
|
MQTT_UPLOAD_FORMAT_TAGS != format.v.val_int) {
|
|
plog_error(plugin, "setting invalid format: %" PRIi64,
|
|
format.v.val_int);
|
|
goto error;
|
|
}
|
|
|
|
// host, required
|
|
if (0 == strlen(host.v.val_str)) {
|
|
plog_error(plugin, "setting invalid host: `%s`", host.v.val_str);
|
|
goto error;
|
|
}
|
|
|
|
if (AUTH_SAS != auth.v.val_int && AUTH_CERT != auth.v.val_int) {
|
|
plog_error(plugin, "setting invalid auth %" PRIu64, auth.v.val_int);
|
|
goto error;
|
|
}
|
|
|
|
// password
|
|
if (AUTH_SAS == auth.v.val_int) {
|
|
ret = neu_parse_param(setting, NULL, 1, &password);
|
|
if (0 != ret) {
|
|
plog_error(plugin, "setting no password");
|
|
goto error;
|
|
}
|
|
} else {
|
|
password.v.val_str = strdup("");
|
|
}
|
|
|
|
// ca, required for TLS (SSL is always enabled)
|
|
ret = neu_parse_param(setting, NULL, 1, &ca);
|
|
if (0 != ret) {
|
|
plog_error(plugin, "setting no ca");
|
|
goto error;
|
|
}
|
|
if (0 != decode_b64_param(plugin, &ca)) {
|
|
goto error;
|
|
}
|
|
|
|
if (AUTH_CERT == auth.v.val_int) {
|
|
ret = parse_ssl_params(plugin, setting, &cert, &key);
|
|
if (0 != ret) {
|
|
plog_error(plugin, "setting certificates fail");
|
|
goto error;
|
|
}
|
|
}
|
|
|
|
// write request topic
|
|
if (0 > neu_asprintf(&write_req_topic, "devices/%s/messages/devicebound/#",
|
|
client_id.v.val_str)) {
|
|
plog_error(plugin, "setting write request topic error");
|
|
goto error;
|
|
}
|
|
|
|
// write response topic
|
|
if (0 > neu_asprintf(&write_resp_topic, "devices/%s/messages/events/",
|
|
client_id.v.val_str)) {
|
|
plog_error(plugin, "setting write response topic error");
|
|
goto error;
|
|
}
|
|
|
|
if (0 > neu_asprintf(&username, "%s/%s/?api-version=2021-04-12",
|
|
host.v.val_str, client_id.v.val_str)) {
|
|
plog_error(plugin, "gen username fail");
|
|
goto error;
|
|
}
|
|
|
|
config->version = NEU_MQTT_VERSION_V311;
|
|
config->client_id = client_id.v.val_str;
|
|
config->qos = qos.v.val_int;
|
|
config->format = format.v.val_int;
|
|
config->write_req_topic = write_req_topic;
|
|
config->write_resp_topic = write_resp_topic;
|
|
config->cache = false;
|
|
config->cache_mem_size = 0;
|
|
config->cache_disk_size = 0;
|
|
config->cache_sync_interval = NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT;
|
|
config->host = host.v.val_str;
|
|
config->port = 8883;
|
|
config->username = username;
|
|
config->password = password.v.val_str;
|
|
config->ssl = true;
|
|
config->ca = ca.v.val_str;
|
|
config->cert = cert.v.val_str;
|
|
config->key = key.v.val_str;
|
|
config->upload_drv_state = false;
|
|
config->enable_topic = true;
|
|
|
|
plog_notice(plugin, "config MQTT version : %d", config->version);
|
|
plog_notice(plugin, "config client-id : %s", config->client_id);
|
|
plog_notice(plugin, "config qos : %d", config->qos);
|
|
plog_notice(plugin, "config format : %s",
|
|
mqtt_upload_format_str(config->format));
|
|
plog_notice(plugin, "config write-req-topic : %s", config->write_req_topic);
|
|
plog_notice(plugin, "config write-resp-topic: %s",
|
|
config->write_resp_topic);
|
|
plog_notice(plugin, "config upload-drv-state: %d",
|
|
config->upload_drv_state);
|
|
plog_notice(plugin, "config cache : %zu", config->cache);
|
|
plog_notice(plugin, "config cache-mem-size : %zu", config->cache_mem_size);
|
|
plog_notice(plugin, "config cache-disk-size : %zu",
|
|
config->cache_disk_size);
|
|
plog_notice(plugin, "config cache-sync-interval : %zu",
|
|
config->cache_sync_interval);
|
|
plog_notice(plugin, "config host : %s", config->host);
|
|
plog_notice(plugin, "config port : %" PRIu16, config->port);
|
|
|
|
if (config->username) {
|
|
plog_notice(plugin, "config username : %s", config->username);
|
|
}
|
|
if (config->password) {
|
|
plog_notice(plugin, "config password : %s",
|
|
0 == strlen(config->password) ? "" : placeholder);
|
|
}
|
|
|
|
plog_notice(plugin, "config ssl : %d", config->ssl);
|
|
if (config->ca) {
|
|
plog_notice(plugin, "config ca : %s", placeholder);
|
|
}
|
|
if (config->cert) {
|
|
plog_notice(plugin, "config cert : %s", placeholder);
|
|
}
|
|
if (config->key) {
|
|
plog_notice(plugin, "config key : %s", placeholder);
|
|
}
|
|
|
|
return 0;
|
|
|
|
error:
|
|
free(err_param);
|
|
free(client_id.v.val_str);
|
|
free(write_req_topic);
|
|
free(write_resp_topic);
|
|
free(host.v.val_str);
|
|
free(username);
|
|
free(password.v.val_str);
|
|
free(ca.v.val_str);
|
|
free(cert.v.val_str);
|
|
free(key.v.val_str);
|
|
return -1;
|
|
}
|
|
|
|
static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
|
|
{
|
|
// write request topic
|
|
if (NULL == plugin->upload_topic ||
|
|
0 != strstr(plugin->upload_topic, config->client_id)) {
|
|
free(plugin->upload_topic);
|
|
plugin->upload_topic = NULL;
|
|
neu_asprintf(&plugin->upload_topic, "devices/%s/messages/events/",
|
|
config->client_id);
|
|
if (NULL == plugin->upload_topic) {
|
|
plog_error(plugin, "create upload topic fail");
|
|
return NEU_ERR_EINTERNAL;
|
|
}
|
|
}
|
|
|
|
if (0 !=
|
|
neu_mqtt_client_subscribe(plugin->client, config->qos,
|
|
config->write_req_topic, plugin,
|
|
handle_write_req)) {
|
|
plog_error(plugin, "subscribe [%s] fail",
|
|
plugin->config.write_req_topic);
|
|
return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int unsubscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
|
|
{
|
|
neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic);
|
|
neu_msleep(100); // wait for message completion
|
|
return 0;
|
|
}
|
|
|
|
static neu_plugin_t *azure_plugin_open(void)
|
|
{
|
|
neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
|
|
neu_plugin_common_init(&plugin->common);
|
|
plugin->parse_config = azure_parse_config;
|
|
plugin->subscribe = subscribe;
|
|
plugin->unsubscribe = unsubscribe;
|
|
|
|
return plugin;
|
|
}
|
|
|
|
static int azure_handle_trans_data(neu_plugin_t * plugin,
|
|
neu_reqresp_trans_data_t *trans_data)
|
|
{
|
|
int rv = 0;
|
|
|
|
if (NULL == plugin->client) {
|
|
return NEU_ERR_MQTT_IS_NULL;
|
|
}
|
|
|
|
if (0 == plugin->config.cache &&
|
|
!neu_mqtt_client_is_connected(plugin->client)) {
|
|
// cache disable and we are disconnected
|
|
return NEU_ERR_MQTT_FAILURE;
|
|
}
|
|
|
|
char *json_str = generate_upload_json(
|
|
plugin, trans_data, plugin->config.format, NULL, 0, NULL, 0, NULL);
|
|
if (NULL == json_str) {
|
|
plog_error(plugin, "generate upload json fail");
|
|
return NEU_ERR_EINTERNAL;
|
|
}
|
|
|
|
char * topic = plugin->upload_topic;
|
|
neu_mqtt_qos_e qos = plugin->config.qos;
|
|
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
|
|
json_str = NULL;
|
|
|
|
return rv;
|
|
}
|
|
|
|
static int azure_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
|
|
void *data)
|
|
{
|
|
neu_err_code_e error = NEU_ERR_SUCCESS;
|
|
|
|
// update cached messages number per seconds
|
|
if (NULL != plugin->client &&
|
|
(global_timestamp - plugin->cache_metric_update_ts) >= 1000) {
|
|
NEU_PLUGIN_UPDATE_METRIC(
|
|
plugin, NEU_METRIC_CACHED_MSGS_NUM,
|
|
neu_mqtt_client_get_cached_msgs_num(plugin->client), NULL);
|
|
plugin->cache_metric_update_ts = global_timestamp;
|
|
}
|
|
|
|
switch (head->type) {
|
|
case NEU_RESP_ERROR:
|
|
error =
|
|
handle_write_response(plugin, head->ctx, data, NULL, NULL, NULL);
|
|
break;
|
|
case NEU_RESP_READ_GROUP:
|
|
// error = handle_read_response(plugin, head->ctx, data);
|
|
break;
|
|
case NEU_REQRESP_TRANS_DATA: {
|
|
if (plugin->client && neu_mqtt_client_is_open(plugin->client)) {
|
|
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL);
|
|
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1,
|
|
NULL);
|
|
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1,
|
|
NULL);
|
|
}
|
|
error = azure_handle_trans_data(plugin, data);
|
|
break;
|
|
}
|
|
case NEU_REQ_SUBSCRIBE_GROUP:
|
|
case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
|
|
neu_req_subscribe_t *sub_info = data;
|
|
free(sub_info->params);
|
|
break;
|
|
}
|
|
case NEU_REQ_UNSUBSCRIBE_GROUP:
|
|
case NEU_REQ_UPDATE_GROUP:
|
|
case NEU_REQ_DEL_GROUP:
|
|
case NEU_REQ_UPDATE_NODE:
|
|
case NEU_REQRESP_NODE_DELETED:
|
|
break;
|
|
default:
|
|
error = NEU_ERR_MQTT_FAILURE;
|
|
break;
|
|
}
|
|
|
|
return error;
|
|
}
|
|
|
|
const neu_plugin_intf_funs_t azure_plugin_intf_funs = {
|
|
.open = azure_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 = azure_plugin_request,
|
|
};
|
|
|
|
#define DESCRIPTION "Northbound plugin for connecting to Azure IoT Hub"
|
|
#define DESCRIPTION_ZH "基于 NanoSDK 的北向应用插件，用于对接 Azure IoT Hub"
|
|
|
|
const neu_plugin_module_t neu_plugin_module = {
|
|
.version = NEURON_PLUGIN_VER_1_0,
|
|
.schema = "azure-iot",
|
|
.module_name = "Azure IoT",
|
|
.module_descr = DESCRIPTION,
|
|
.module_descr_zh = DESCRIPTION_ZH,
|
|
.intf_funs = &azure_plugin_intf_funs,
|
|
.kind = NEU_PLUGIN_KIND_SYSTEM,
|
|
.type = NEU_NA_TYPE_APP,
|
|
.display = true,
|
|
.single = false,
|
|
};
|