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

Browse Source 
feat: add mqtt client
This commit is contained in:
fengzeroz
2022-11-14 17:06:04 +08:00
committed by GitHub
parent 8cfc5c9cd7 4ca5b5ad37
commit 9402cc6d52
11 changed files with 586 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1 -3
View File
@@ -66,8 +66,7 @@ set(NEURON_BASE_SOURCES
src/base/tag_sort.c
src/base/group.c
src/connection/connection.c
src/connection/mqtt_c_client.c
src/connection/mqtt_client_intf.c
src/connection/mqtt_client.c
src/event/event_linux.c
src/event/event_unix.c
src/utils/asprintf.c
@@ -81,7 +80,6 @@ add_library(neuron-base SHARED)
target_sources(neuron-base PRIVATE ${NEURON_BASE_SOURCES} ${NEURON_SRC_PARSE})
target_include_directories(neuron-base
PRIVATE include/neuron src)
target_compile_definitions(neuron-base PRIVATE MQTT_USE_BIO)
target_link_libraries(neuron-base libssl.a libcrypto.a)
target_link_libraries(neuron-base nng libzlog.so jansson mqttc jwt
${CMAKE_THREAD_LIBS_INIT})
include/neuron/connection/mqtt_client.h
+82
View File
@@ -0,0 +1,82 @@
/**
* 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.
**/
#ifndef NEURON_CONNECTION_MQTT_CLIENT_H
#define NEURON_CONNECTION_MQTT_CLIENT_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include "utils/zlog.h"
typedef void (*neu_mqtt_client_cb_t)(void *data);
typedef enum {
NEU_MQTT_VERSION_V31 = 3,
NEU_MQTT_VERSION_V311 = 4,
NEU_MQTT_VERSION_V5 = 5,
} neu_mqtt_version_e;
typedef enum {
MQTT_QOS0,
MQTT_QOS1,
MQTT_QOS2,
} neu_mqtt_qos_e;
typedef struct neu_mqtt_client_s neu_mqtt_client_t;
neu_mqtt_client_t *neu_mqtt_client_new(const char *host, uint16_t port,
neu_mqtt_version_e version);
void neu_mqtt_client_free(neu_mqtt_client_t *client);
bool neu_mqtt_client_is_opened(neu_mqtt_client_t *client);
bool neu_mqtt_client_is_connected(neu_mqtt_client_t *client);
int neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username,
const char *password);
int neu_mqtt_client_set_connect_cb(neu_mqtt_client_t * client,
neu_mqtt_client_cb_t cb, void *data);
int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t * client,
neu_mqtt_client_cb_t cb, void *data);
int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
const char *cert, const char *key,
const char *keypass);
int neu_mqtt_client_set_msg_cache_limit(neu_mqtt_client_t *client,
size_t cache_limit);
int neu_mqtt_client_set_zlog_category(neu_mqtt_client_t *client,
zlog_category_t * cat);
int neu_mqtt_client_open(neu_mqtt_client_t *client);
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);
#ifdef __cplusplus
}
#endif
#endif
include/neuron/neuron.h
-1
View File
@@ -45,7 +45,6 @@ extern "C" {
#include "utils/uthash.h"
#include "utils/utlist.h"
#include "connection/mqtt_client_intf.h"
#include "connection/neu_connection.h"
#include "event/event.h"
plugins/mqtt/CMakeLists.txt
+3
View File
@@ -7,6 +7,8 @@ set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins")
file(COPY ${CMAKE_SOURCE_DIR}/plugins/mqtt/mqtt.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
set(src
mqtt_c_client.c
mqtt_client_intf.c
mqtt_plugin.c
mqtt_util.c
command/read_write.c
@@ -19,3 +21,4 @@ target_include_directories(${PROJECT_NAME} PRIVATE
${CMAKE_SOURCE_DIR}/plugins/mqtt)
target_link_libraries(${PROJECT_NAME} neuron-base)
target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT})
target_compile_definitions(${PROJECT_NAME} PRIVATE MQTT_USE_BIO)
src/connection/mqtt_c_client.c → plugins/mqtt/mqtt_c_client.c
+6 -6
View File
@@ -71,10 +71,10 @@ struct reconnect_state {
};
struct subscribe_tuple {
char * topic;
int qos;
neu_subscribe_handle handle;
mqtt_c_client_t * client;
char * topic;
int qos;
subscribe_handle handle;
mqtt_c_client_t *client;
};
struct mqtt_c_client {
@@ -845,7 +845,7 @@ static neu_err_code_e client_destroy(mqtt_c_client_t *client)
static struct subscribe_tuple *
client_subscribe_create(mqtt_c_client_t *client, const char *topic,
const int qos, const neu_subscribe_handle handle)
const int qos, const subscribe_handle handle)
{
struct subscribe_tuple *tuple = calloc(1, sizeof(struct subscribe_tuple));
if (NULL == tuple) {
@@ -883,7 +883,7 @@ static neu_err_code_e client_subscribe_add(mqtt_c_client_t * client,
neu_err_code_e mqtt_c_client_subscribe(mqtt_c_client_t *client,
const char *topic, const int qos,
neu_subscribe_handle handle)
subscribe_handle handle)
{
assert(NULL != client);
assert(NULL != topic && 0 < strlen(topic));
src/connection/mqtt_c_client.h → plugins/mqtt/mqtt_c_client.h
+4 -3
View File
@@ -24,10 +24,11 @@
extern "C" {
#endif
#include "connection/mqtt_client_intf.h"
#include <stddef.h>
#include <stdint.h>
#include "mqtt_client_intf.h"
typedef struct mqtt_c_client mqtt_c_client_t;
neu_err_code_e mqtt_c_client_open(mqtt_c_client_t ** p_client,
@@ -36,7 +37,7 @@ neu_err_code_e mqtt_c_client_open(mqtt_c_client_t ** p_client,
neu_err_code_e mqtt_c_client_is_connected(mqtt_c_client_t *client);
neu_err_code_e mqtt_c_client_subscribe(mqtt_c_client_t *client,
const char *topic, const int qos,
neu_subscribe_handle handle);
subscribe_handle handle);
neu_err_code_e mqtt_c_client_unsubscribe(mqtt_c_client_t *client,
const char * topic);
neu_err_code_e mqtt_c_client_publish(mqtt_c_client_t *client, const char *topic,
@@ -49,4 +50,4 @@ neu_err_code_e mqtt_c_client_close(mqtt_c_client_t *client);
#ifdef __cplusplus
}
#endif
#endif
#endif
src/connection/mqtt_client_intf.c → plugins/mqtt/mqtt_client_intf.c
+12 -16
View File
@@ -17,52 +17,48 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include "connection/mqtt_client_intf.h"
#include "mqtt_client_intf.h"
#include "mqtt_c_client.h"
neu_err_code_e neu_mqtt_client_open(neu_mqtt_client_t * p_client,
const neu_mqtt_option_t *option,
void * context)
neu_err_code_e mqtt_client_open(mqtt_client_t * p_client,
const neu_mqtt_option_t *option, void *context)
{
return mqtt_c_client_open((mqtt_c_client_t **) p_client, option, context);
}
neu_err_code_e neu_mqtt_client_is_connected(neu_mqtt_client_t client)
neu_err_code_e mqtt_client_is_connected(mqtt_client_t client)
{
return mqtt_c_client_is_connected(client);
}
neu_err_code_e neu_mqtt_client_subscribe(neu_mqtt_client_t client,
const char *topic, const int qos,
neu_subscribe_handle handle)
neu_err_code_e mqtt_client_subscribe(mqtt_client_t client, const char *topic,
const int qos, subscribe_handle handle)
{
return mqtt_c_client_subscribe(client, topic, qos, handle);
}
neu_err_code_e neu_mqtt_client_unsubscribe(neu_mqtt_client_t client,
const char * topic)
neu_err_code_e mqtt_client_unsubscribe(mqtt_client_t client, const char *topic)
{
return mqtt_c_client_unsubscribe(client, topic);
}
neu_err_code_e neu_mqtt_client_publish(neu_mqtt_client_t client,
const char *topic, int qos,
unsigned char *payload, size_t len)
neu_err_code_e mqtt_client_publish(mqtt_client_t client, const char *topic,
int qos, unsigned char *payload, size_t len)
{
return mqtt_c_client_publish(client, topic, qos, payload, len);
}
neu_err_code_e neu_mqtt_client_suspend(neu_mqtt_client_t client)
neu_err_code_e mqtt_client_suspend(mqtt_client_t client)
{
return mqtt_c_client_suspend(client);
}
neu_err_code_e neu_mqtt_client_continue(neu_mqtt_client_t client)
neu_err_code_e mqtt_client_continue(mqtt_client_t client)
{
return mqtt_c_client_continue(client);
}
neu_err_code_e neu_mqtt_client_close(neu_mqtt_client_t client)
neu_err_code_e mqtt_client_close(mqtt_client_t client)
{
return mqtt_c_client_close(client);
}
include/neuron/connection/mqtt_client_intf.h → plugins/mqtt/mqtt_client_intf.h
+16 -20
View File
@@ -67,36 +67,32 @@ typedef struct {
zlog_category_t *log;
} neu_mqtt_option_t;
typedef void *neu_mqtt_client_t;
typedef void *mqtt_client_t;
typedef void (*neu_subscribe_handle)(const char *topic_name, size_t topic_len,
void *payload, const size_t len,
void *context);
typedef void (*subscribe_handle)(const char *topic_name, size_t topic_len,
void *payload, const size_t len,
void *context);
neu_err_code_e neu_mqtt_client_open(neu_mqtt_client_t * p_client,
const neu_mqtt_option_t *option,
void * context);
neu_err_code_e mqtt_client_open(mqtt_client_t * p_client,
const neu_mqtt_option_t *option, void *context);
neu_err_code_e neu_mqtt_client_is_connected(neu_mqtt_client_t client);
neu_err_code_e mqtt_client_is_connected(mqtt_client_t client);
neu_err_code_e neu_mqtt_client_subscribe(neu_mqtt_client_t client,
const char *topic, const int qos,
neu_subscribe_handle handle);
neu_err_code_e mqtt_client_subscribe(mqtt_client_t client, const char *topic,
const int qos, subscribe_handle handle);
neu_err_code_e neu_mqtt_client_unsubscribe(neu_mqtt_client_t client,
const char * topic);
neu_err_code_e mqtt_client_unsubscribe(mqtt_client_t client, const char *topic);
neu_err_code_e neu_mqtt_client_publish(neu_mqtt_client_t client,
const char *topic, int qos,
unsigned char *payload, size_t len);
neu_err_code_e mqtt_client_publish(mqtt_client_t client, const char *topic,
int qos, unsigned char *payload, size_t len);
neu_err_code_e neu_mqtt_client_suspend(neu_mqtt_client_t client);
neu_err_code_e mqtt_client_suspend(mqtt_client_t client);
neu_err_code_e neu_mqtt_client_continue(neu_mqtt_client_t client);
neu_err_code_e mqtt_client_continue(mqtt_client_t client);
neu_err_code_e neu_mqtt_client_close(neu_mqtt_client_t client);
neu_err_code_e mqtt_client_close(mqtt_client_t client);
#ifdef __cplusplus
}
#endif
#endif
#endif
plugins/mqtt/mqtt_plugin.c
+21 -22
View File
@@ -52,7 +52,7 @@ struct topic_pair {
struct mqtt_routine {
neu_plugin_t * plugin;
neu_mqtt_option_t option;
neu_mqtt_client_t client;
mqtt_client_t client;
UT_array * topics;
};
@@ -181,16 +181,15 @@ static void mqtt_routine_response(const char *topic_name, size_t topic_len,
command_response_handle(&response);
}
static void topics_subscribe(UT_array *topics, neu_mqtt_client_t client)
static void topics_subscribe(UT_array *topics, 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);
mqtt_client_subscribe(client, (*pair)->topic_request,
(*pair)->qos_request, mqtt_routine_response);
}
}
}
@@ -231,16 +230,16 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin,
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);
mqtt_client_t *client_p = (mqtt_client_t *) &routine->client;
neu_err_code_e error = NEU_ERR_SUCCESS;
error = 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);
mqtt_client_continue(routine->client);
plog_info(plugin, "open mqtt client: %s:%s, code:%d", routine->option.host,
routine->option.port, error);
@@ -260,10 +259,10 @@ static void mqtt_routine_stop(mqtt_routine_t *routine)
assert(NULL != routine);
// stop recevied
neu_mqtt_client_suspend(routine->client);
mqtt_client_suspend(routine->client);
// close mqtt client and clean
neu_mqtt_client_close(routine->client);
mqtt_client_close(routine->client);
plog_info(routine->plugin, "close mqtt client:%s:%s", routine->option.host,
routine->option.port);
@@ -448,8 +447,8 @@ static neu_err_code_e write_response(neu_plugin_t * plugin,
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));
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);
@@ -479,8 +478,8 @@ static neu_err_code_e read_response(neu_plugin_t * plugin,
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));
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);
@@ -515,8 +514,8 @@ static int cache_send(neu_plugin_t *plugin)
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));
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);
@@ -544,7 +543,7 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
return NEU_ERR_MQTT_FAILURE;
}
neu_err_code_e rc = neu_mqtt_client_is_connected(routine->client);
neu_err_code_e rc = mqtt_client_is_connected(routine->client);
if (0 != rc) {
cache_item_t item = {
.size = strlen(json_str) + 1,
@@ -573,8 +572,8 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *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));
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);
@@ -610,8 +609,8 @@ static neu_err_code_e node_state_send(neu_plugin_t * plugin,
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));
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);
plugins/mqtt/mqtt_util.h
+2 -1
View File
@@ -24,9 +24,10 @@
extern "C" {
#endif
#include "connection/mqtt_client_intf.h"
#include <neuron.h>
#include "mqtt_client_intf.h"
int mqtt_option_init(neu_plugin_t *plugin, char *config,
neu_mqtt_option_t *option);
src/connection/mqtt_client.c
+439
View File
@@ -0,0 +1,439 @@
/**
* 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 <inttypes.h>
#include <stdlib.h>
#include <string.h>
#define NNG_SUPP_SQLITE 1
#include <nng/mqtt/mqtt_client.h>
#include <nng/nng.h>
#include <nng/supplemental/tls/tls.h>
#include <nng/supplemental/util/platform.h>
#include "connection/mqtt_client.h"
#include "utils/asprintf.h"
#include "utils/zlog.h"
#define log(level, ...) \
do { \
if (client->log) { \
zlog_##level(client->log, ##__VA_ARGS__); \
} \
} while (0)
#define check_opened() \
do { \
if (client->opened) { \
log(warn, \
"mqtt client already opened, setting will not take effect"); \
return -1; \
} \
} while (0)
struct neu_mqtt_client_s {
nng_socket sock;
char * url;
neu_mqtt_version_e version;
nng_tls_config * tls_cfg;
nng_msg * conn_msg;
bool opened;
bool connected;
neu_mqtt_client_cb_t connect_cb;
void * connect_cb_data;
neu_mqtt_client_cb_t disconnect_cb;
void * disconnect_cb_data;
nng_mqtt_sqlite_option *sqlite_cfg;
zlog_category_t * log;
};
static inline uint8_t neu_mqtt_version_to_nng_mqtt_version(neu_mqtt_version_e v)
{
switch (v) {
case NEU_MQTT_VERSION_V31:
return MQTT_PROTOCOL_VERSION_v31;
case NEU_MQTT_VERSION_V311:
return MQTT_PROTOCOL_VERSION_v311;
case NEU_MQTT_VERSION_V5:
return MQTT_PROTOCOL_VERSION_v5;
default:
assert(!"logic error, invalid mqtt version");
}
}
static void connect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
{
(void) p;
(void) ev;
neu_mqtt_client_t *client = arg;
log(info, "mqtt client connected");
client->connected = true;
if (client->connect_cb) {
client->connect_cb(client->connect_cb_data);
}
}
static void disconnect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
{
(void) p;
(void) ev;
neu_mqtt_client_t *client = arg;
log(info, "mqtt client disconnected");
client->connected = false;
if (client->disconnect_cb) {
client->disconnect_cb(client->connect_cb_data);
}
}
static nng_msg *alloc_conn_msg(neu_mqtt_client_t *client,
neu_mqtt_version_e version)
{
int rv = 0;
nng_msg *msg = NULL;
if (0 != (rv = nng_mqtt_msg_alloc(&msg, 0))) {
log(error, "nng_mqtt_msg_alloc fail: %s", nng_strerror(rv));
return NULL;
}
nng_mqtt_msg_set_packet_type(msg, NNG_MQTT_CONNECT);
nng_mqtt_msg_set_connect_keep_alive(msg, 60);
nng_mqtt_msg_set_connect_proto_version(
msg, neu_mqtt_version_to_nng_mqtt_version(version));
nng_mqtt_msg_set_connect_clean_session(msg, true);
return msg;
}
static inline nng_tls_config *alloc_tls_config(neu_mqtt_client_t *client)
{
int rv = 0;
nng_tls_config *cfg = NULL;
if ((rv = nng_tls_config_alloc(&cfg, NNG_TLS_MODE_CLIENT)) != 0) {
log(error, "nng_tls_config_alloc fail: %s", nng_strerror(rv));
return NULL;
}
return cfg;
}
static inline nng_mqtt_sqlite_option *
alloc_sqlite_config(neu_mqtt_client_t *client)
{
int rv;
nng_mqtt_sqlite_option *cfg = NULL;
if ((rv = nng_mqtt_alloc_sqlite_opt(&cfg)) != 0) {
log(error, "nng_mqtt_alloc_sqlite_opt fail: %s", nng_strerror(rv));
return NULL;
}
nng_mqtt_set_sqlite_enable(cfg, true);
nng_mqtt_set_sqlite_flush_threshold(cfg, 256);
nng_mqtt_set_sqlite_max_rows(cfg, 256);
nng_mqtt_set_sqlite_db_dir(cfg, "persistence/");
nng_mqtt_sqlite_db_init(cfg, "mqtt_client.db", client->version);
return cfg;
}
neu_mqtt_client_t *neu_mqtt_client_new(const char *host, uint16_t port,
neu_mqtt_version_e version)
{
char *url = NULL;
neu_asprintf(&url, "mqtt-tcp://%s:%" PRIu16, host, port);
if (NULL == url) {
return NULL;
}
neu_mqtt_client_t *client = calloc(1, sizeof(*client));
if (NULL == client) {
free(url);
return NULL;
}
client->conn_msg = alloc_conn_msg(client, version);
if (NULL == client->conn_msg) {
free(url);
return NULL;
}
client->url = url;
client->version = version;
return client;
}
void neu_mqtt_client_free(neu_mqtt_client_t *client)
{
if (client) {
if (client->tls_cfg) {
nng_tls_config_free(client->tls_cfg);
}
if (client->sqlite_cfg) {
nng_mqtt_free_sqlite_opt(client->sqlite_cfg);
}
nng_msg_free(client->conn_msg);
free(client->url);
free(client);
}
}
bool neu_mqtt_client_is_opened(neu_mqtt_client_t *client)
{
return client->opened;
}
bool neu_mqtt_client_is_connected(neu_mqtt_client_t *client)
{
return client->connected;
}
int neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username,
const char *password)
{
check_opened();
nng_mqtt_msg_set_connect_user_name(client->conn_msg, username);
nng_mqtt_msg_set_connect_password(client->conn_msg, password);
return 0;
}
int neu_mqtt_client_set_connect_cb(neu_mqtt_client_t * client,
neu_mqtt_client_cb_t cb, void *data)
{
check_opened();
client->connect_cb = cb;
client->connect_cb_data = data;
return 0;
}
int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t * client,
neu_mqtt_client_cb_t cb, void *data)
{
check_opened();
client->disconnect_cb = cb;
client->disconnect_cb_data = data;
return 0;
}
int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
const char *cert, const char *key,
const char *keypass)
{
int rv = 0;
nng_tls_config *cfg = NULL;
check_opened();
if (client->tls_cfg) {
cfg = client->tls_cfg;
} else {
cfg = alloc_tls_config(client);
if (NULL == cfg) {
return -1;
}
client->tls_cfg = cfg;
}
if (cert != NULL && key != NULL) {
if ((rv = nng_tls_config_auth_mode(cfg, NNG_TLS_AUTH_MODE_REQUIRED)) !=
0) {
log(error, "nng_tls_config_auth_mode fail: %s", nng_strerror(rv));
return -1;
}
if ((rv = nng_tls_config_own_cert(cfg, cert, key, keypass)) != 0) {
log(error, "nng_tls_config_own_cert fail: %s", nng_strerror(rv));
return -1;
}
} else {
if ((rv = nng_tls_config_auth_mode(cfg, NNG_TLS_AUTH_MODE_NONE)) != 0) {
log(error, "nng_tls_config_auth_mode fail: %s", nng_strerror(rv));
return -1;
}
}
if (ca != NULL) {
if ((rv = nng_tls_config_ca_chain(cfg, ca, NULL)) != 0) {
log(error, "nng_tls_config_ca_chain fail: %s", nng_strerror(rv));
return -1;
}
}
return 0;
}
int neu_mqtt_client_set_msg_cache_limit(neu_mqtt_client_t *client,
size_t cache_limit)
{
check_opened();
if (NULL == client->sqlite_cfg) {
client->sqlite_cfg = alloc_sqlite_config(client);
if (NULL == client->sqlite_cfg) {
return -1;
}
}
nng_mqtt_set_sqlite_max_rows(client->sqlite_cfg, cache_limit);
return 0;
}
int neu_mqtt_client_set_zlog_category(neu_mqtt_client_t *client,
zlog_category_t * cat)
{
check_opened();
client->log = cat;
return 0;
}
int neu_mqtt_client_open(neu_mqtt_client_t *client)
{
int rv = 0;
if (client->opened) {
return rv;
}
if (NEU_MQTT_VERSION_V5 == client->version &&
(rv = nng_mqttv5_client_open(&client->sock)) != 0) {
log(error, "nng_mqttv5_client_open fail: %s", nng_strerror(rv));
return -1;
} else if ((rv = nng_mqtt_client_open(&client->sock)) != 0) {
log(error, "nng_mqtt_client_open fail: %s", nng_strerror(rv));
return -1;
}
if (client->sqlite_cfg &&
(rv = nng_socket_set_ptr(client->sock, NNG_OPT_MQTT_SQLITE,
client->sqlite_cfg))) {
log(error, "nng_socket_set_ptr(NNG_OPT_MQTT_SQLITE) fail: %s",
nng_strerror(rv));
return -1;
}
nng_dialer dialer;
if ((rv = nng_dialer_create(&dialer, client->sock, client->url)) != 0) {
log(error, "nng_dialer_create fail: %s", nng_strerror(rv));
return -1;
}
if (client->tls_cfg &&
0 !=
(rv = nng_dialer_set_ptr(dialer, NNG_OPT_TLS_CONFIG,
client->tls_cfg))) {
log(error, "nng_dialter_set_ptr(NNG_OPT_TLS_CONFIG) fail: %s",
nng_strerror(rv));
return -1;
}
nng_mqtt_set_connect_cb(client->sock, connect_cb, client);
nng_mqtt_set_disconnect_cb(client->sock, disconnect_cb, client);
nng_dialer_set_ptr(dialer, NNG_OPT_MQTT_CONNMSG, client->conn_msg);
nng_dialer_start(dialer, NNG_FLAG_NONBLOCK);
client->opened = true;
return 0;
}
int neu_mqtt_client_close(neu_mqtt_client_t *client)
{
int rv = 0;
if (!client->opened) {
return rv;
}
rv = nng_close(client->sock);
if (0 != rv) {
log(error, "nng_close: %s", nng_strerror(rv));
return -1;
}
client->opened = false;
return 0;
}
int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
const char *topic, const char *payload, size_t len)
{
nng_msg *pub_msg;
nng_mqtt_msg_alloc(&pub_msg, 0);
nng_mqtt_msg_set_packet_type(pub_msg, NNG_MQTT_PUBLISH);
nng_mqtt_msg_set_publish_topic(pub_msg, topic);
nng_mqtt_msg_set_publish_payload(pub_msg, (uint8_t *) payload, len);
nng_mqtt_msg_set_publish_qos(pub_msg, qos);
int rv = nng_sendmsg(client->sock, pub_msg, NNG_FLAG_NONBLOCK);
if (0 != rv) {
log(error, "nng_sendmsg fail: %s", nng_strerror(rv));
return -1;
}
log(debug, "pub [%s, QoS%d] %zu bytes", topic, qos, len);
return 0;
}
int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
const char *topic, neu_mqtt_client_cb_t cb)
{
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_qos topic_qos[] = {
{ .qos = qos,
.topic = { .buf = (uint8_t *) topic, .length = strlen(topic) } },
};
size_t topic_qos_count = sizeof(topic_qos) / sizeof(nng_mqtt_topic_qos);
nng_mqtt_msg_set_packet_type(sub_msg, NNG_MQTT_SUBSCRIBE);
nng_mqtt_msg_set_subscribe_topics(sub_msg, topic_qos, topic_qos_count);
rv = nng_sendmsg(client->sock, sub_msg, NNG_FLAG_NONBLOCK);
if (0 != rv) {
log(error, "nng_sendmsg fail: %s", nng_strerror(rv));
return -1;
}
(void) cb; // TODO: handle callback on subscribed topic
return 0;
}