diff --git a/CMakeLists.txt b/CMakeLists.txt index f7ac91c5..c2782b78 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -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}) diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h new file mode 100644 index 00000000..8e5a270a --- /dev/null +++ b/include/neuron/connection/mqtt_client.h @@ -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 +#include +#include + +#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 diff --git a/include/neuron/neuron.h b/include/neuron/neuron.h index 38764e84..c3ec8b30 100644 --- a/include/neuron/neuron.h +++ b/include/neuron/neuron.h @@ -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" diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index d89dfd99..771a496b 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -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) diff --git a/src/connection/mqtt_c_client.c b/plugins/mqtt/mqtt_c_client.c similarity index 99% rename from src/connection/mqtt_c_client.c rename to plugins/mqtt/mqtt_c_client.c index d469c3e5..315f15f2 100644 --- a/src/connection/mqtt_c_client.c +++ b/plugins/mqtt/mqtt_c_client.c @@ -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)); diff --git a/src/connection/mqtt_c_client.h b/plugins/mqtt/mqtt_c_client.h similarity index 94% rename from src/connection/mqtt_c_client.h rename to plugins/mqtt/mqtt_c_client.h index 4f160173..d7299ecf 100644 --- a/src/connection/mqtt_c_client.h +++ b/plugins/mqtt/mqtt_c_client.h @@ -24,10 +24,11 @@ extern "C" { #endif -#include "connection/mqtt_client_intf.h" #include #include +#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 \ No newline at end of file +#endif diff --git a/src/connection/mqtt_client_intf.c b/plugins/mqtt/mqtt_client_intf.c similarity index 56% rename from src/connection/mqtt_client_intf.c rename to plugins/mqtt/mqtt_client_intf.c index edba1997..efaa7e81 100644 --- a/src/connection/mqtt_client_intf.c +++ b/plugins/mqtt/mqtt_client_intf.c @@ -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); } diff --git a/include/neuron/connection/mqtt_client_intf.h b/plugins/mqtt/mqtt_client_intf.h similarity index 60% rename from include/neuron/connection/mqtt_client_intf.h rename to plugins/mqtt/mqtt_client_intf.h index 786e3a44..2217cbdf 100644 --- a/include/neuron/connection/mqtt_client_intf.h +++ b/plugins/mqtt/mqtt_client_intf.h @@ -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 \ No newline at end of file +#endif diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index c2d3b06c..6595ad21 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -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); diff --git a/plugins/mqtt/mqtt_util.h b/plugins/mqtt/mqtt_util.h index 45c72211..edd2d303 100644 --- a/plugins/mqtt/mqtt_util.h +++ b/plugins/mqtt/mqtt_util.h @@ -24,9 +24,10 @@ extern "C" { #endif -#include "connection/mqtt_client_intf.h" #include +#include "mqtt_client_intf.h" + int mqtt_option_init(neu_plugin_t *plugin, char *config, neu_mqtt_option_t *option); diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c new file mode 100644 index 00000000..947cbb65 --- /dev/null +++ b/src/connection/mqtt_client.c @@ -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 +#include +#include +#include + +#define NNG_SUPP_SQLITE 1 +#include +#include +#include +#include + +#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; +}