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

Merge pull request #1240 from eeff/eeff/dev

Browse Source 
feat(mqtt): add write request/response topic parameters
This commit is contained in:
eeff
2023-05-06 14:35:02 +08:00
committed by GitHub
parent 8484edb44f bc6cb180e6
commit 3f2ecd86fc
9 changed files with 319 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client.h
+2
View File
@@ -114,6 +114,8 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
const char *topic, void *data,
neu_mqtt_client_subscribe_cb_t cb);
int neu_mqtt_client_unsubscribe(neu_mqtt_client_t *client, const char *topic);
#ifdef __cplusplus
}
#endif
include/neuron/utils/time.h
+10
View File
@@ -26,6 +26,7 @@ extern "C" {
#include <stdint.h>
#include <stdlib.h>
#include <sys/time.h>
#include <time.h>
static inline int64_t neu_time_ms()
{
@@ -34,6 +35,15 @@ static inline int64_t neu_time_ms()
return (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000;
}
static inline void neu_msleep(unsigned msec)
{
struct timespec tv = {
.tv_sec = msec / 1000,
.tv_nsec = (msec % 1000) * 1000000,
};
nanosleep(&tv, NULL);
}
#ifdef __cplusplus
}
#endif
plugins/mqtt/mqtt.json
+25 -1
View File
@@ -57,6 +57,30 @@
]
}
},
"write-req-topic": {
"name": "Write Request Topic",
"name_zh": "写请求主题",
"description": "MQTT topic to receive write request messages.",
"description_zh": "接收点位写入请求的 MQTT 主题。",
"attribute": "required",
"type": "string",
"default": "/neuron/${node-name}/write/req",
"valid": {
"length": 255
}
},
"write-resp-topic": {
"name": "Write Response Topic",
"name_zh": "写响应主题",
"description": "MQTT topic to send write response messages.",
"description_zh": "发送点位写入响应的 MQTT 主题。",
"attribute": "required",
"type": "string",
"default": "/neuron/${node-name}/write/resp",
"valid": {
"length": 255
}
},
"offline-cache": {
"name": "Offline Data Caching",
"name_zh": "离线缓存",
@@ -218,4 +242,4 @@
"length": 256
}
}
}
}
plugins/mqtt/mqtt_config.c
+67 -30
View File
@@ -17,6 +17,7 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "utils/asprintf.h"
#include "json/json.h"
#include "json/neu_json_param.h"
@@ -203,6 +204,18 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT };
neu_json_elem_t write_req_topic = {
.name = "write-req-topic",
.t = NEU_JSON_STR,
.v.val_str = NULL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t write_resp_topic = {
.name = "write-resp-topic",
.t = NEU_JSON_STR,
.v.val_str = NULL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t offline_cache = { .name = "offline-cache",
.t = NEU_JSON_BOOL };
neu_json_elem_t cache_mem_size = { .name = "cache-mem-size",
@@ -224,8 +237,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
return -1;
}
ret = neu_parse_param(setting, &err_param, 5, &client_id, &format, &qos,
&host, &port);
ret = neu_parse_param(setting, &err_param, 7, &client_id, &qos, &format,
&write_req_topic, &write_resp_topic, &host, &port);
if (0 != ret) {
plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
goto error;
@@ -251,6 +264,22 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
// write request topic
if (NULL == write_req_topic.v.val_str &&
0 > neu_asprintf(&write_req_topic.v.val_str, "/neuron/%s/write/req",
plugin->common.name)) {
plog_error(plugin, "setting write request topic error");
goto error;
}
// write response topic
if (NULL == write_resp_topic.v.val_str &&
0 > neu_asprintf(&write_resp_topic.v.val_str, "/neuron/%s/write/resp",
plugin->common.name)) {
plog_error(plugin, "setting write request topic error");
goto error;
}
// offline cache
ret = parse_cache_params(plugin, setting, &offline_cache, &cache_mem_size,
&cache_disk_size);
@@ -287,55 +316,61 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->cache = offline_cache.v.val_bool;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
config->cache_disk_size = cache_disk_size.v.val_int * MB;
config->host = host.v.val_str;
config->port = port.v.val_int;
config->username = username.v.val_str;
config->password = password.v.val_str;
config->ca = ca.v.val_str;
config->cert = cert.v.val_str;
config->key = key.v.val_str;
config->keypass = keypass.v.val_str;
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.v.val_str;
config->write_resp_topic = write_resp_topic.v.val_str;
config->cache = offline_cache.v.val_bool;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
config->cache_disk_size = cache_disk_size.v.val_int * MB;
config->host = host.v.val_str;
config->port = port.v.val_int;
config->username = username.v.val_str;
config->password = password.v.val_str;
config->ca = ca.v.val_str;
config->cert = cert.v.val_str;
config->key = key.v.val_str;
config->keypass = keypass.v.val_str;
plog_info(plugin, "config client-id : %s", config->client_id);
plog_info(plugin, "config qos : %d", config->qos);
plog_info(plugin, "config format : %s",
plog_info(plugin, "config client-id : %s", config->client_id);
plog_info(plugin, "config qos : %d", config->qos);
plog_info(plugin, "config format : %s",
mqtt_upload_format_str(config->format));
plog_info(plugin, "config cache : %zu", config->cache);
plog_info(plugin, "config cache-mem-size : %zu", config->cache_mem_size);
plog_info(plugin, "config cache-disk-size: %zu", config->cache_disk_size);
plog_info(plugin, "config host : %s", config->host);
plog_info(plugin, "config port : %" PRIu16, config->port);
plog_info(plugin, "config write-req-topic : %s", config->write_req_topic);
plog_info(plugin, "config write-resp-topic: %s", config->write_req_topic);
plog_info(plugin, "config cache : %zu", config->cache);
plog_info(plugin, "config cache-mem-size : %zu", config->cache_mem_size);
plog_info(plugin, "config cache-disk-size : %zu", config->cache_disk_size);
plog_info(plugin, "config host : %s", config->host);
plog_info(plugin, "config port : %" PRIu16, config->port);
if (config->username) {
plog_info(plugin, "config username : %s", config->username);
plog_info(plugin, "config username : %s", config->username);
}
if (config->password) {
plog_info(plugin, "config password : %s",
plog_info(plugin, "config password : %s",
0 == strlen(config->password) ? "" : placeholder);
}
if (config->ca) {
plog_info(plugin, "config ca : %s", placeholder);
plog_info(plugin, "config ca : %s", placeholder);
}
if (config->cert) {
plog_info(plugin, "config cert : %s", placeholder);
plog_info(plugin, "config cert : %s", placeholder);
}
if (config->key) {
plog_info(plugin, "config key : %s", placeholder);
plog_info(plugin, "config key : %s", placeholder);
}
if (config->keypass) {
plog_info(plugin, "config keypass : %s", placeholder);
plog_info(plugin, "config keypass : %s", placeholder);
}
return 0;
error:
free(err_param);
free(write_req_topic.v.val_str);
free(write_resp_topic.v.val_str);
free(host.v.val_str);
free(username.v.val_str);
free(password.v.val_str);
@@ -349,6 +384,8 @@ error:
void mqtt_config_fini(mqtt_config_t *config)
{
free(config->client_id);
free(config->write_req_topic);
free(config->write_resp_topic);
free(config->host);
free(config->username);
free(config->password);
plugins/mqtt/mqtt_config.h
+16 -14
View File
@@ -48,20 +48,22 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
}
typedef struct {
char * client_id; // client id
mqtt_upload_format_e format; // upload format
neu_mqtt_qos_e qos; // message QoS
size_t cache; // cache enable flag
size_t cache_mem_size; // cache memory size in bytes
size_t cache_disk_size; // cache disk size in bytes
char * host; // broker host
uint16_t port; // broker port
char * username; // user name
char * password; // user password
char * ca; // CA
char * cert; // client cert
char * key; // client key
char * keypass; // client key password
char * client_id; // client id
neu_mqtt_qos_e qos; // message QoS
mqtt_upload_format_e format; // upload format
char * write_req_topic; // write request topic
char * write_resp_topic; // write response topic
size_t cache; // cache enable flag
size_t cache_mem_size; // cache memory size in bytes
size_t cache_disk_size; // cache disk size in bytes
char * host; // broker host
uint16_t port; // broker port
char * username; // user name
char * password; // user password
char * ca; // CA
char * cert; // client cert
char * key; // client key
char * keypass; // client key password
} mqtt_config_t;
int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
plugins/mqtt/mqtt_handle.c
+1 -1
View File
@@ -373,7 +373,7 @@ int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
goto end;
}
char * topic = plugin->write_resp_topic;
char * topic = plugin->config.write_resp_topic;
neu_mqtt_qos_e qos = plugin->config.qos;
rv = publish(plugin, qos, topic, json_str, strlen(json_str));
json_str = NULL;
plugins/mqtt/mqtt_plugin.c
+73 -78
View File
@@ -18,6 +18,7 @@
**/
#include "utils/asprintf.h"
#include "utils/time.h"
#include "mqtt_config.h"
#include "mqtt_handle.h"
@@ -81,10 +82,6 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin)
plugin->read_req_topic = NULL;
free(plugin->read_resp_topic);
plugin->read_resp_topic = NULL;
free(plugin->write_req_topic);
plugin->write_req_topic = NULL;
free(plugin->write_resp_topic);
plugin->write_resp_topic = NULL;
route_tbl_free(plugin->route_tbl);
@@ -159,6 +156,65 @@ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
return rv;
}
static int create_topic(neu_plugin_t *plugin)
{
if (plugin->read_req_topic) {
// topics already created
return 0;
}
neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
plugin->common.name);
if (NULL == plugin->read_req_topic) {
return -1;
}
neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
plugin->common.name);
if (NULL == plugin->read_resp_topic) {
free(plugin->read_req_topic);
plugin->read_req_topic = NULL;
return -1;
}
return 0;
}
static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
{
if (0 != create_topic(plugin)) {
plog_error(plugin, "create topics fail");
return NEU_ERR_EINTERNAL;
}
if (0 !=
neu_mqtt_client_subscribe(plugin->client, config->qos,
plugin->read_req_topic, plugin,
handle_read_req)) {
plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
}
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, plugin->read_req_topic);
neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic);
neu_msleep(100); // wait for message completion
return 0;
}
static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
{
int rv = 0;
@@ -181,7 +237,8 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
}
} else if (neu_mqtt_client_is_open(plugin->client)) {
started = true;
rv = neu_mqtt_client_close(plugin->client);
unsubscribe(plugin, &plugin->config);
rv = neu_mqtt_client_close(plugin->client);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_close fail");
rv = NEU_ERR_EINTERNAL;
@@ -195,10 +252,15 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
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 (started) {
if (0 != neu_mqtt_client_open(plugin->client)) {
plog_error(plugin, "neu_mqtt_client_open fail");
rv = NEU_ERR_MQTT_CONNECT_FAILURE;
goto error;
}
if (0 != (rv = subscribe(plugin, &config))) {
goto error;
}
}
if (plugin->config.host) {
@@ -216,52 +278,6 @@ error:
return rv;
}
static int create_topic(neu_plugin_t *plugin)
{
if (plugin->read_req_topic) {
// topics already created
return 0;
}
neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
plugin->common.name);
if (NULL == plugin->read_req_topic) {
return -1;
}
neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
plugin->common.name);
if (NULL == plugin->read_resp_topic) {
free(plugin->read_req_topic);
plugin->read_req_topic = NULL;
return -1;
}
neu_asprintf(&plugin->write_req_topic, "/neuron/%s/write/req",
plugin->common.name);
if (NULL == plugin->write_req_topic) {
free(plugin->read_req_topic);
plugin->read_req_topic = NULL;
free(plugin->read_resp_topic);
plugin->read_resp_topic = NULL;
return -1;
}
neu_asprintf(&plugin->write_resp_topic, "/neuron/%s/write/resp",
plugin->common.name);
if (NULL == plugin->write_resp_topic) {
free(plugin->read_req_topic);
plugin->read_req_topic = NULL;
free(plugin->read_resp_topic);
plugin->read_resp_topic = NULL;
free(plugin->write_req_topic);
plugin->write_req_topic = NULL;
return -1;
}
return 0;
}
static int mqtt_plugin_start(neu_plugin_t *plugin)
{
int rv = 0;
@@ -279,29 +295,7 @@ static int mqtt_plugin_start(neu_plugin_t *plugin)
goto end;
}
if (0 != create_topic(plugin)) {
plog_error(plugin, "create topics fail");
rv = NEU_ERR_EINTERNAL;
goto end;
}
if (0 !=
neu_mqtt_client_subscribe(plugin->client, plugin->config.qos,
plugin->read_req_topic, plugin,
handle_read_req)) {
plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
goto end;
}
if (0 !=
neu_mqtt_client_subscribe(plugin->client, plugin->config.qos,
plugin->write_req_topic, plugin,
handle_write_req)) {
plog_error(plugin, "subscribe [%s] fail", plugin->write_req_topic);
rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
goto end;
}
rv = subscribe(plugin, &plugin->config);
end:
if (0 == rv) {
@@ -317,6 +311,7 @@ end:
static int mqtt_plugin_stop(neu_plugin_t *plugin)
{
if (plugin->client) {
unsubscribe(plugin, &plugin->config);
neu_mqtt_client_close(plugin->client);
plog_info(plugin, "mqtt client closed");
}
plugins/mqtt/mqtt_plugin.h
-2
View File
@@ -49,8 +49,6 @@ struct neu_plugin {
int64_t cache_metric_update_ts;
char * read_req_topic;
char * read_resp_topic;
char * write_req_topic;
char * write_resp_topic;
route_entry_t * route_tbl;
};
src/connection/mqtt_client.c
+125 -3
View File
@@ -33,6 +33,7 @@
#include "connection/mqtt_client.h"
#include "event/event.h"
#include "utils/asprintf.h"
#include "utils/time.h"
#include "utils/utextend.h"
#include "utils/uthash.h"
#include "utils/utlist.h"
@@ -68,6 +69,7 @@ typedef struct {
typedef enum {
TASK_PUB,
TASK_SUB,
TASK_UNSUB,
TASK_RECV,
} task_kind_e;
@@ -133,7 +135,8 @@ static inline void tasks_free(task_t *tasks);
static void task_cb(void *arg);
static void task_handle_pub(task_t *task, neu_mqtt_client_t *client);
static void task_handle_sub(task_t *task, neu_mqtt_client_t *client);
static void task_handle_recv(task_t *task, neu_mqtt_client_t *client);
static void task_handle_unsub(task_t *task, neu_mqtt_client_t *client);
static void task_handle_recv(task_t *task, neu_mqtt_client_t *client);
static subscription_t * subscription_new(neu_mqtt_client_t *client,
neu_mqtt_qos_e qos, const char *topic,
@@ -221,6 +224,8 @@ static void task_cb(void *arg)
task_handle_pub(task, client);
} else if (TASK_SUB == task->kind) {
task_handle_sub(task, client);
} else if (TASK_UNSUB == task->kind) {
task_handle_unsub(task, client);
} else if (TASK_RECV == task->kind) {
task_handle_recv(task, client);
} else {
@@ -295,6 +300,41 @@ end:
return;
}
static void task_handle_unsub(task_t *task, neu_mqtt_client_t *client)
{
int rv = 0;
nng_msg *msg = NULL;
nng_aio *aio = task->aio;
if (0 != (rv = nng_aio_result(aio))) {
log(error, "send UNSUBSCRIBE error: %s", nng_strerror(rv));
goto end;
}
msg = nng_aio_get_msg(aio);
if (NULL == msg) {
rv = -1;
goto end;
}
nng_mqtt_packet_type pt = nng_mqtt_msg_get_packet_type(msg);
if (NNG_MQTT_UNSUBACK != pt) {
log(error, "recv packet type:%d, expect UNSUBACK packet", pt);
rv = -1;
goto end;
}
end:
if (task->sub) {
log(info, "unsub [%s, QoS%d], %s", task->sub->topic, task->sub->qos,
(0 == rv) ? "OK" : "FAIL");
} else {
log(info, "unsub %s", (0 == rv) ? "OK" : "FAIL");
}
nng_msg_free(msg);
return;
}
static void task_handle_recv(task_t *task, neu_mqtt_client_t *client)
{
(void) client;
@@ -535,7 +575,7 @@ static inline void client_free_task(neu_mqtt_client_t *client, task_t *task)
{
nng_aio_set_msg(task->aio, NULL);
if (TASK_SUB == task->kind) {
if (TASK_SUB == task->kind || TASK_UNSUB == task->kind) {
subscription_free(task->sub);
} else if (TASK_RECV == task->kind) {
subscription_free(task->recv.sub);
@@ -569,6 +609,16 @@ static inline void client_add_subscription(neu_mqtt_client_t *client,
HASH_ADD_STR(client->subscriptions, topic, sub);
}
static inline void client_del_subscription(neu_mqtt_client_t *client,
subscription_t * sub)
{
HASH_DEL(client->subscriptions, sub);
if (sub->ack) {
client->suback_count -= 1;
}
subscription_free(sub);
}
static int client_send_sub_msg(neu_mqtt_client_t *client,
subscription_t * subscription)
{
@@ -606,6 +656,38 @@ static int client_send_sub_msg(neu_mqtt_client_t *client,
return 0;
}
static int client_send_unsub_msg(neu_mqtt_client_t *client,
subscription_t * subscription)
{
int rv = 0;
nng_msg *sub_msg = NULL;
if ((rv = nng_mqtt_msg_alloc(&sub_msg, 0)) != 0) {
log(error, "nng_mqtt_msg_alloc fail: %s", nng_strerror(rv));
return -1;
}
nng_mqtt_topic mqtt_topic = { .buf = (uint8_t *) subscription->topic,
.length = strlen(subscription->topic) };
nng_mqtt_msg_set_packet_type(sub_msg, NNG_MQTT_UNSUBSCRIBE);
nng_mqtt_msg_set_unsubscribe_topics(sub_msg, &mqtt_topic, 1);
task_t *task = client_alloc_task(client);
if (NULL == task) {
log(error, "client_alloc_task fail");
nng_msg_free(sub_msg);
return -1;
}
task->kind = TASK_UNSUB;
task->sub = subscription ? subscription_ref(subscription) : NULL;
nng_aio_set_msg(task->aio, sub_msg);
nng_send_aio(client->sock, task->aio);
return 0;
}
static inline void client_start_recv(neu_mqtt_client_t *client)
{
if (client->recv_aio && !client->receiving) {
@@ -1163,7 +1245,7 @@ int neu_mqtt_client_close(neu_mqtt_client_t *client)
// wait for all tasks
while (client_task_free_list_len(client) != client->task_count) {
nng_mtx_unlock(client->mtx);
nng_msleep(100);
neu_msleep(100);
nng_mtx_lock(client->mtx);
}
client->open = false;
@@ -1261,3 +1343,43 @@ error:
subscription_free(subscription);
return -1;
}
int neu_mqtt_client_unsubscribe(neu_mqtt_client_t *client, const char *topic)
{
int rv = 0;
subscription_t *subscription = NULL;
nng_mtx_lock(client->mtx);
if (NULL == client->recv_aio) {
if ((rv = nng_aio_alloc(&client->recv_aio, recv_cb, client)) != 0) {
log(error, "nng_aio_alloc fail: %s", nng_strerror(rv));
goto error;
}
if (client->connected) {
// connect_cb already fired, then start receiving
client_start_recv(client);
}
}
HASH_FIND_STR(client->subscriptions, topic, subscription);
if (client->connected) {
// send a unsubscribe message even subscription is NULL
if (client_send_unsub_msg(client, subscription) < 0) {
log(error, "client_send_unsub_msg fail");
goto error;
}
}
if (subscription) {
client_del_subscription(client, subscription);
}
nng_mtx_unlock(client->mtx);
return 0;
error:
nng_mtx_unlock(client->mtx);
return -1;
}