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 #916 from eeff/eeff/dev

Browse Source 
feat(nano-mqtt): handle read/write request
This commit is contained in:
fengzeroz
2022-11-16 09:32:28 +08:00
committed by GitHub
parent d74a91c9c1 5862b3e528
commit ef536f6f99
6 changed files with 490 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client.h
+4 -1
View File
@@ -31,6 +31,8 @@ extern "C" {
#include "utils/zlog.h"
typedef void (*neu_mqtt_client_cb_t)(void *data);
typedef void (*neu_mqtt_client_subscribe_cb_t)(const uint8_t *payload,
uint32_t len, void *data);
typedef enum {
NEU_MQTT_VERSION_V31 = 3,
@@ -73,7 +75,8 @@ int neu_mqtt_client_close(neu_mqtt_client_t *client);
int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
const char *topic, const char *payload, size_t len);
int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
const char *topic, neu_mqtt_client_cb_t cb);
const char * topic,
neu_mqtt_client_subscribe_cb_t cb, void *data);
#ifdef __cplusplus
}
plugins/nano-mqtt/mqtt_handle.c
+254 -8
View File
@@ -17,8 +17,11 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "connection/mqtt_client.h"
#include "errcodes.h"
#include "version.h"
#include "json/neu_json_mqtt.h"
#include "json/neu_json_rw.h"
#include "mqtt_handle.h"
#include "mqtt_plugin.h"
@@ -176,22 +179,265 @@ static char *generate_upload_json(neu_plugin_t * plugin,
return json_str;
}
static char *generate_read_resp_json(neu_plugin_t * plugin,
neu_json_mqtt_t * mqtt,
neu_resp_read_group_t *data)
{
neu_resp_tag_value_t *tags = data->tags;
uint16_t len = data->n_tag;
char * json_str = NULL;
neu_json_read_resp_t json = { 0 };
if (0 != tag_values_to_json(tags, len, &json)) {
plog_error(plugin, "tag_values_to_json fail");
return NULL;
}
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, mqtt,
neu_json_encode_mqtt_resp, &json_str);
if (json.tags) {
free(json.tags);
}
return json_str;
}
static char *generate_write_resp_json(neu_plugin_t * plugin,
neu_json_mqtt_t * mqtt,
neu_resp_error_t *data)
{
(void) plugin;
neu_json_error_resp_t error = { .error = data->error };
char * json_str = NULL;
neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, mqtt,
neu_json_encode_mqtt_resp, &json_str);
return json_str;
}
static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req)
{
plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
req->node);
neu_reqresp_head_t header = { 0 };
header.ctx = mqtt;
header.type = NEU_REQ_READ_GROUP;
neu_req_read_group_t cmd = { 0 };
strcpy(cmd.driver, req->node);
strcpy(cmd.group, req->group);
if (0 != neu_plugin_op(plugin, header, &cmd)) {
plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail");
return -1;
}
return 0;
}
static int send_write_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_write_req_t *req)
{
plog_info(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
req->group, req->node);
neu_reqresp_head_t header = { 0 };
neu_req_write_tag_t cmd = { 0 };
header.ctx = mqtt;
header.type = NEU_REQ_WRITE_TAG;
strcpy(cmd.driver, req->node);
strcpy(cmd.group, req->group);
strcpy(cmd.tag, req->tag);
switch (req->t) {
case NEU_JSON_INT:
cmd.value.type = NEU_TYPE_INT64;
cmd.value.value.u64 = req->value.val_int;
break;
case NEU_JSON_STR:
cmd.value.type = NEU_TYPE_STRING;
strcpy(cmd.value.value.str, req->value.val_str);
break;
case NEU_JSON_DOUBLE:
cmd.value.type = NEU_TYPE_DOUBLE;
cmd.value.value.d64 = req->value.val_double;
break;
case NEU_JSON_BOOL:
cmd.value.type = NEU_TYPE_BOOL;
cmd.value.value.boolean = req->value.val_bool;
break;
default:
plog_error(plugin, "invalid tag value type: %d", req->t);
return -1;
}
if (0 != neu_plugin_op(plugin, header, &cmd)) {
plog_error(plugin, "neu_plugin_op(NEU_REQ_WRITE_TAG) fail");
return -1;
}
return 0;
}
void handle_write_req(const uint8_t *payload, uint32_t len, void *data)
{
int rv = 0;
neu_plugin_t * plugin = data;
neu_json_write_req_t *req = NULL;
char *json_str = malloc(len + 1);
if (NULL == json_str) {
return;
}
memcpy(json_str, payload, len);
json_str[len] = '\0';
neu_json_mqtt_t *mqtt = NULL;
rv = neu_json_decode_mqtt_req(json_str, &mqtt);
if (0 != rv) {
plog_error(plugin, "neu_json_decode_mqtt_req failed");
free(json_str);
return;
}
rv = neu_json_decode_write_req(json_str, &req);
if (0 != rv) {
plog_error(plugin, "neu_json_decode_write_req fail");
neu_json_decode_mqtt_req_free(mqtt);
free(json_str);
return;
}
rv = send_write_req(plugin, mqtt, req);
if (0 != rv) {
neu_json_decode_mqtt_req_free(mqtt);
}
neu_json_decode_write_req_free(req);
free(json_str);
}
int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_error_t *data)
{
(void) plugin;
(void) mqtt_json;
(void) data;
return 0;
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_write_resp_json(plugin, mqtt_json, data);
if (NULL == json_str) {
plog_error(plugin, "generate write resp json fail, uuid:%s",
mqtt_json->uuid);
rv = NEU_ERR_EINTERNAL;
goto end;
}
const char * topic = plugin->write_resp_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, "pub [%s, QoS%d] error", topic, qos);
rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
}
end:
free(json_str);
neu_json_decode_mqtt_req_free(mqtt_json);
return rv;
}
void handle_read_req(const uint8_t *payload, uint32_t len, void *data)
{
int rv = 0;
neu_plugin_t * plugin = data;
neu_json_read_req_t *req = NULL;
char *json_str = malloc(len + 1);
if (NULL == json_str) {
return;
}
memcpy(json_str, payload, len);
json_str[len] = '\0';
neu_json_mqtt_t *mqtt = NULL;
rv = neu_json_decode_mqtt_req(json_str, &mqtt);
if (0 != rv) {
plog_error(plugin, "neu_json_decode_mqtt_req failed");
free(json_str);
return;
}
rv = neu_json_decode_read_req(json_str, &req);
if (0 != rv) {
plog_error(plugin, "neu_json_decode_read_req fail");
neu_json_decode_mqtt_req_free(mqtt);
free(json_str);
return;
}
rv = send_read_req(plugin, mqtt, req);
if (0 != rv) {
neu_json_decode_mqtt_req_free(mqtt);
}
neu_json_decode_read_req_free(req);
free(json_str);
}
int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_read_group_t *data)
{
(void) plugin;
(void) mqtt_json;
(void) data;
return 0;
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_read_resp_json(plugin, mqtt_json, data);
if (NULL == json_str) {
plog_error(plugin, "generate read resp json fail");
rv = NEU_ERR_EINTERNAL;
goto end;
}
const char * topic = plugin->read_resp_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, "pub [%s, QoS%d] error", topic, qos);
rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
goto end;
}
end:
neu_json_decode_mqtt_req_free(mqtt_json);
free(json_str);
free(data->tags);
return rv;
}
int handle_trans_data(neu_plugin_t * plugin,
plugins/nano-mqtt/mqtt_handle.h
+4
View File
@@ -26,9 +26,13 @@ extern "C" {
#include "neuron.h"
void handle_write_req(const uint8_t *payload, uint32_t len, void *data);
int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_error_t *data);
void handle_read_req(const uint8_t *payload, uint32_t len, void *data);
int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_read_group_t *data);
plugins/nano-mqtt/mqtt_plugin.c
+82 -1
View File
@@ -17,9 +17,11 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "mqtt_plugin.h"
#include "utils/asprintf.h"
#include "mqtt_config.h"
#include "mqtt_handle.h"
#include "mqtt_plugin.h"
const neu_plugin_module_t neu_plugin_module;
@@ -69,6 +71,16 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin)
neu_mqtt_client_free(plugin->client);
plugin->client = NULL;
}
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;
free(plugin->write_resp_topic);
plugin->write_resp_topic = NULL;
plog_info(plugin, "uninitialize plugin `%s` success",
neu_plugin_module.module_name);
return NEU_ERR_SUCCESS;
@@ -185,6 +197,52 @@ 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->read_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->read_req_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;
@@ -202,11 +260,34 @@ 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, 0, plugin->read_req_topic,
handle_read_req, plugin)) {
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, 0, plugin->write_req_topic,
handle_write_req, plugin)) {
plog_error(plugin, "subscribe [%s] fail", plugin->write_req_topic);
rv = NEU_ERR_MQTT_SUBSCRIBE_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);
neu_mqtt_client_close(plugin->client);
}
return rv;
}
plugins/nano-mqtt/mqtt_plugin.h
+4
View File
@@ -33,6 +33,10 @@ struct neu_plugin {
neu_plugin_common_t common;
mqtt_config_t config;
neu_mqtt_client_t * client;
char * read_req_topic;
char * read_resp_topic;
char * write_req_topic;
char * write_resp_topic;
};
#ifdef __cplusplus
src/connection/mqtt_client.c
+142 -5
View File
@@ -30,6 +30,8 @@
#include "connection/mqtt_client.h"
#include "utils/asprintf.h"
#include "utils/utextend.h"
#include "utils/uthash.h"
#include "utils/zlog.h"
#define log(level, ...) \
@@ -48,6 +50,14 @@
} \
} while (0)
typedef struct {
char * topic;
neu_mqtt_client_subscribe_cb_t cb;
void * data;
nng_aio * aio;
UT_hash_handle hh;
} subscription_t;
struct neu_mqtt_client_s {
nng_socket sock;
char * url;
@@ -61,6 +71,7 @@ struct neu_mqtt_client_s {
neu_mqtt_client_cb_t disconnect_cb;
void * disconnect_cb_data;
nng_mqtt_sqlite_option *sqlite_cfg;
subscription_t * subscriptions;
zlog_category_t * log;
};
@@ -106,6 +117,44 @@ static void disconnect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
}
}
static void recv_cb(void *arg)
{
int rv = 0;
subscription_t * subscription = arg;
nng_aio * aio = subscription->aio;
neu_mqtt_client_t *client = nng_aio_get_input(aio, 0);
if (0 != (rv = nng_aio_result(aio))) {
log(error, "mqtt client recv error: %s", nng_strerror(rv));
if (NNG_EAGAIN == rv) {
nng_recv_aio(client->sock, aio);
}
return;
}
nng_msg *msg = nng_aio_get_msg(aio);
uint32_t payload_len;
uint8_t * payload = nng_mqtt_msg_get_publish_payload(msg, &payload_len);
uint32_t topic_len;
const char *topic = nng_mqtt_msg_get_publish_topic(msg, &topic_len);
uint8_t qos = nng_mqtt_msg_get_publish_qos(msg);
log(info, "recv [%.*s, QoS%d] %" PRIu32 " bytes", (unsigned) topic_len,
topic, (int) qos, payload_len);
// find matching subscription
HASH_FIND(hh, client->subscriptions, topic, topic_len, subscription);
if (NULL != subscription) {
subscription->cb(payload, payload_len, subscription->data);
} else {
log(warn, "[%.*s] no subscription found", (unsigned) topic_len, topic);
}
nng_msg_free(msg);
nng_recv_aio(client->sock, aio);
}
static nng_msg *alloc_conn_msg(neu_mqtt_client_t *client,
neu_mqtt_version_e version)
{
@@ -159,6 +208,78 @@ alloc_sqlite_config(neu_mqtt_client_t *client)
return cfg;
}
static subscription_t *subscription_new(neu_mqtt_client_t * client,
const char * topic,
neu_mqtt_client_subscribe_cb_t cb,
void * data)
{
int rv = 0;
subscription_t *subscription = NULL;
subscription = calloc(1, sizeof(*subscription));
if (NULL == subscription) {
return NULL;
}
subscription->topic = strdup(topic);
if (NULL == subscription->topic) {
free(subscription);
return NULL;
}
if ((rv = nng_aio_alloc(&subscription->aio, recv_cb, subscription)) != 0) {
log(error, "nng_aio_alloc fail: %s", nng_strerror(rv));
free(subscription->topic);
free(subscription);
return NULL;
}
if ((rv = nng_aio_set_input(subscription->aio, 0, client)) != 0) {
log(error, "nng_aio_set_input fail: %s", nng_strerror(rv));
nng_aio_free(subscription->aio);
free(subscription->topic);
free(subscription);
return NULL;
}
subscription->cb = cb;
subscription->data = data;
return subscription;
}
static inline void subscription_free(subscription_t *subscription)
{
if (subscription) {
nng_aio_stop(subscription->aio);
nng_aio_free(subscription->aio);
free(subscription->topic);
free(subscription);
}
}
static inline void subscriptions_free(subscription_t *subscriptions)
{
subscription_t *sub = NULL, *tmp = NULL;
HASH_ITER(hh, subscriptions, sub, tmp)
{
HASH_DEL(subscriptions, sub);
subscription_free(sub);
}
}
static void add_subscription(neu_mqtt_client_t *client, subscription_t *sub)
{
subscription_t *old = NULL;
HASH_FIND_STR(client->subscriptions, sub->topic, old);
if (old) {
HASH_DEL(client->subscriptions, old);
subscription_free(old);
}
HASH_ADD_STR(client->subscriptions, topic, sub);
}
neu_mqtt_client_t *neu_mqtt_client_new(const char *host, uint16_t port,
neu_mqtt_version_e version)
{
@@ -196,6 +317,7 @@ void neu_mqtt_client_free(neu_mqtt_client_t *client)
if (client->sqlite_cfg) {
nng_mqtt_free_sqlite_opt(client->sqlite_cfg);
}
subscriptions_free(client->subscriptions);
nng_msg_free(client->conn_msg);
free(client->url);
free(client);
@@ -412,16 +534,23 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
}
int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
const char *topic, neu_mqtt_client_cb_t cb)
const char * topic,
neu_mqtt_client_subscribe_cb_t cb, void *data)
{
int rv = 0;
nng_msg *sub_msg = NULL;
int rv = 0;
nng_msg * sub_msg = NULL;
subscription_t *subscription = NULL;
if ((rv = nng_mqtt_msg_alloc(&sub_msg, 0)) != 0) {
log(error, "nng_mqtt_msg_alloc fail: %s", nng_strerror(rv));
return -1;
}
subscription = subscription_new(client, topic, cb, data);
if (NULL == subscription) {
goto error;
}
nng_mqtt_topic_qos topic_qos[] = {
{ .qos = qos,
.topic = { .buf = (uint8_t *) topic, .length = strlen(topic) } },
@@ -435,10 +564,18 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
rv = nng_sendmsg(client->sock, sub_msg, NNG_FLAG_NONBLOCK);
if (0 != rv) {
log(error, "nng_sendmsg fail: %s", nng_strerror(rv));
return -1;
goto error;
}
(void) cb; // TODO: handle callback on subscribed topic
add_subscription(client, subscription);
log(info, "sub [%s, QoS%d]", topic, qos);
nng_recv_aio(client->sock, subscription->aio);
return 0;
error:
nng_msg_free(sub_msg);
subscription_free(subscription);
return -1;
}