diff --git a/README.md b/README.md index cc0f4e7e..f5254fb9 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,18 @@ NEURON IIoT System for Industry 4.0 $ ninja $ ninja install ``` + + [MQTT-C](https://github.com/LiamBindle/MQTT-C.git) + + ```shell + $ git clone -b 1.1.5 https://github.com/LiamBindle/MQTT-C.git + $ cd MQTT-C + $ mkdir build + $ cd build + $ cmake -G Ninja -DCMAKE_POSITION_INDEPENDENT_CODE=ON .. + $ ninja + $ ninja install + ``` - [yaml](https://github.com/yaml/libyaml.git) ```shell diff --git a/install-libs.sh b/install-libs.sh index 1681b4de..f35c00b9 100755 --- a/install-libs.sh +++ b/install-libs.sh @@ -10,7 +10,7 @@ if [ $system == 'Linux' ];then fi fi -lib_list=(nng jansson jwt paho openssl gtest open62541 yaml) +lib_list=(nng jansson jwt paho MQTT-C openssl gtest open62541 yaml) list_str="" for var in ${lib_list[*]};do @@ -102,6 +102,24 @@ function build_paho() cd ../../ } +# MQTT-C +function build_MQTT-C() +{ + echo "Building MQTT-C (1.1.5)" + if [ ! -d MQTT-C ];then + git clone -b 1.1.5 https://github.com/LiamBindle/MQTT-C.git + fi + cd MQTT-C + rm -rf build + mkdir build + cd build + cmake -G Ninja -DCMAKE_POSITION_INDEPENDENT_CODE=ON .. + ninja + ninja install + echo "Leaving MQTT-C " + cd ../../ +} + function build_yaml() { git clone https://github.com/yaml/libyaml.git diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index b308c0be..5df380b1 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -7,7 +7,8 @@ set(src ${PARSER_SRC} ${CONFIG_SRC} plugin_mqtt.c - paho_client.c + # paho_client.c + mqttc_client.c common.c command_rw.c command_group_config.c @@ -20,5 +21,9 @@ target_include_directories(plugin-mqtt PRIVATE ${CMAKE_SOURCE_DIR}/include ) -target_link_libraries(plugin-mqtt neuron-base paho-mqtt3as ssl crypto yaml) +target_link_libraries(plugin-mqtt + neuron-base mqttc + # paho-mqtt3as + ssl crypto + yaml) target_link_libraries(plugin-mqtt ${CMAKE_THREAD_LIBS_INIT}) diff --git a/plugins/mqtt/command_rw.c b/plugins/mqtt/command_rw.c index ac6a9e70..85c4079d 100644 --- a/plugins/mqtt/command_rw.c +++ b/plugins/mqtt/command_rw.c @@ -178,8 +178,6 @@ char *command_read_once_response(neu_plugin_t * plugin, neu_data_val_t * resp_val) { UNUSED(plugin); - UNUSED(resp_val); - UNUSED(parse_header); neu_fixed_array_t *array; int rc = neu_dvalue_get_ref_array(resp_val, &array); @@ -248,7 +246,6 @@ static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id, uint32_t req_id = neu_plugin_get_event_id(plugin); neu_plugin_send_write_cmd(plugin, req_id, dest_node_id, config, write_val); - return req_id; } @@ -360,11 +357,11 @@ static void clean_write_response_json_object(neu_parse_read_res_t *json) free(json->tags); } -char *command_write_response(neu_plugin_t *plugin, const char *uuid, - neu_data_val_t *resp_val) +char *command_write_response(neu_plugin_t * plugin, + neu_parse_mqtt_t *parse_header, + neu_data_val_t * resp_val) { UNUSED(plugin); - UNUSED(uuid); neu_fixed_array_t *array; neu_dvalue_get_ref_array(resp_val, &array); @@ -381,8 +378,8 @@ char *command_write_response(neu_plugin_t *plugin, const char *uuid, wrap_write_response_json_object(array, &json); char *json_str = NULL; - neu_json_encode_by_fn(&json, neu_parse_encode_read, &json_str); + neu_json_encode_with_mqtt(&json, neu_parse_encode_read, parse_header, + neu_parse_encode_mqtt_param, &json_str); clean_write_response_json_object(&json); - return json_str; } \ No newline at end of file diff --git a/plugins/mqtt/command_rw.h b/plugins/mqtt/command_rw.h index b2d9bd09..3c5e2bd2 100644 --- a/plugins/mqtt/command_rw.h +++ b/plugins/mqtt/command_rw.h @@ -37,8 +37,9 @@ char *command_read_cycle_response(neu_plugin_t * plugin, neu_data_val_t *resp_val); int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, neu_parse_write_req_t *write_req); -char *command_write_response(neu_plugin_t *plugin, const char *uuid, - neu_data_val_t *resp_val); +char *command_write_response(neu_plugin_t * plugin, + neu_parse_mqtt_t *parse_header, + neu_data_val_t * resp_val); #ifdef __cplusplus } diff --git a/plugins/mqtt/option.h b/plugins/mqtt/mqtt_client.h similarity index 78% rename from plugins/mqtt/option.h rename to plugins/mqtt/mqtt_client.h index 10ad8d0c..8a84ec8f 100644 --- a/plugins/mqtt/option.h +++ b/plugins/mqtt/mqtt_client.h @@ -17,8 +17,8 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#ifndef NEURON_PLUGIN_PAHO_OPTION -#define NEURON_PLUGIN_PAHO_OPTION +#ifndef NEURON_PLUGIN_MQTT_CLIENT +#define NEURON_PLUGIN_MQTT_CLIENT #ifdef __cplusplus extern "C" { @@ -79,6 +79,23 @@ typedef struct { char *https_proxy; } option_t; +typedef enum { + ClientSuccess = 0, + ClientIsNULL, + ClientConnectFailure, + ClientSubscribeTimeout, + ClientSubscribeListInitialFailure, + ClientSubscribeFailure, + ClientSubscribeAddListRepeat, + ClientSubscribeAddListFailure, + ClientUnsubscribeFailure, + ClientPublishFailure, +} client_error; + +typedef void (*subscribe_handle)(const char *topic_name, size_t topic_len, + void *payload, const size_t len, + void *context); + #ifdef __cplusplus } #endif diff --git a/plugins/mqtt/mqttc_client.c b/plugins/mqtt/mqttc_client.c new file mode 100644 index 00000000..1e2a1c86 --- /dev/null +++ b/plugins/mqtt/mqttc_client.c @@ -0,0 +1,396 @@ + +/** + * @file + * A simple subscriber program that performs automatic reconnections. + */ +#include +#include +#include + +#include +#include + +#include "mqttc_client.h" +#include "posix_sockets.h" + +#define UNUSED(x) (void) (x) +#define INTERVAL 100000U +#define SEND_BUF_SIZE 2048 +#define RECV_BUF_SIZE 2048 + +struct reconnect_state_t { + const char * hostname; + const char * port; + const char * client_id; + const char * topic; + mqttc_client_t *client; + uint8_t * send_buf; + size_t send_buf_sz; + uint8_t * recv_buf; + size_t recv_buf_sz; +}; + +struct mqttc_client { + option_t * option; + neu_list subscribe_list; + struct mqtt_client mqttc; + struct reconnect_state_t reconnect_state; + uint8_t send_buf[SEND_BUF_SIZE]; + uint8_t recv_buf[RECV_BUF_SIZE]; + pthread_t client_daemon; + void * user_data; +}; + +struct subscribe_tuple { + neu_list_node node; + char * topic; + int qos; + subscribe_handle handle; + bool subscribed; + mqttc_client_t * client; +}; + +client_error mqttc_client_subscribe_send(mqttc_client_t * client, + struct subscribe_tuple *tuple); + +static void subscribe_all_topic(mqttc_client_t *client) +{ + struct subscribe_tuple *item; + NEU_LIST_FOREACH(&client->subscribe_list, item) + { + mqttc_client_subscribe_send(client, item); + } +} + +static void reconnect_client(struct mqtt_client *client, + void ** reconnect_state_vptr) +{ + struct reconnect_state_t *reconnect_state = + *((struct reconnect_state_t **) reconnect_state_vptr); + + if (client->error != MQTT_ERROR_INITIAL_RECONNECT) { + close(client->socketfd); + } + + if (client->error != MQTT_ERROR_INITIAL_RECONNECT) { + log_error( + "reconnect_client: called while client was in error state \"%s\"", + mqtt_error_str(client->error)); + } + + int sockfd = + open_nb_socket(reconnect_state->hostname, reconnect_state->port); + if (sockfd == -1) { + log_info("Failed to open socket: "); + } + + mqtt_reinit(client, sockfd, reconnect_state->send_buf, + reconnect_state->send_buf_sz, reconnect_state->recv_buf, + reconnect_state->recv_buf_sz); + + /* Create an anonymous session */ + // const char *client_id = NULL; + const char *client_id = reconnect_state->client_id; + /* Ensure we have a clean session */ + uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION; + /* Send connection request to the broker. */ + mqtt_connect(client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, + 400); + + /* Subscribe to the topic. */ + // mqtt_subscribe(client, reconnect_state->topic, 0); + subscribe_all_topic(reconnect_state->client); +} + +static void publish_callback(void ** reconnect_state_vptr, + struct mqtt_response_publish *published) +{ + /* note that published->topic_name is NOT null-terminated (here we'll change + * it to a c-string) */ + char *topic_name = (char *) malloc(published->topic_name_size + 1); + memcpy(topic_name, published->topic_name, published->topic_name_size); + topic_name[published->topic_name_size] = '\0'; + + log_info("Received publish('%s'): %s", topic_name, + (const char *) published->application_message); + + struct reconnect_state_t *reconnect_state = + *((struct reconnect_state_t **) reconnect_state_vptr); + mqttc_client_t * client = reconnect_state->client; + struct subscribe_tuple *item; + NEU_LIST_FOREACH(&client->subscribe_list, item) + { + if (0 == strcmp(item->topic, topic_name) && NULL != item->handle) { + item->handle(topic_name, strlen(topic_name), + (void *) published->application_message, + published->application_message_size, + client->user_data); + } + } + + free(topic_name); +} + +static void *client_refresher(void *client) +{ + while (1) { + mqtt_sync((struct mqtt_client *) client); + usleep(INTERVAL); + } + return NULL; +} + +mqttc_client_t *mqttc_client_create(option_t *option, void *context) +{ + mqttc_client_t *client = malloc(sizeof(mqttc_client_t)); + if (NULL == client) { + log_error("Failed to malloc client memory"); + return NULL; + } + memset(client, 0, sizeof(mqttc_client_t)); + + client->option = option; + client->user_data = context; + + struct reconnect_state_t *reconnect_state = &client->reconnect_state; + reconnect_state->hostname = client->option->host; + reconnect_state->port = client->option->port; + reconnect_state->client_id = client->option->clientid; + reconnect_state->topic = client->option->topic; + reconnect_state->client = client; + reconnect_state->send_buf = client->send_buf; + reconnect_state->send_buf_sz = sizeof(client->send_buf); + reconnect_state->recv_buf = client->recv_buf; + reconnect_state->recv_buf_sz = sizeof(client->recv_buf); + + client->mqttc.publish_response_callback_state = reconnect_state; + + NEU_LIST_INIT(&client->subscribe_list, struct subscribe_tuple, node); + return client; +} + +client_error mqttc_client_connect(mqttc_client_t *client) +{ + if (NULL == client) { + log_error("The client is not initialized"); + return ClientIsNULL; + } + + mqtt_init_reconnect(&client->mqttc, reconnect_client, + &client->reconnect_state, publish_callback); + + if (pthread_create(&client->client_daemon, NULL, client_refresher, + &client->mqttc)) { + log_error("Failed to start client daemon."); + return ClientConnectFailure; + } + + log_info("Successfully start the request of connect."); + return ClientSuccess; +} + +client_error mqttc_client_open(option_t *option, void *context, + mqttc_client_t **p_client) +{ + mqttc_client_t *client = mqttc_client_create(option, context); + if (NULL == client) { + return ClientIsNULL; + } + + client_error rc = mqttc_client_connect(client); + *p_client = client; + return rc; +} + +client_error mqttc_client_disconnect(mqttc_client_t *client) +{ + struct subscribe_tuple *item; + NEU_LIST_FOREACH(&client->subscribe_list, item) + { + client_error error = mqttc_client_unsubscribe(client, item->topic); + if (ClientSuccess != error) { + // TODO: Error handle + continue; + } + } + + if (NULL != client) { + int rc = mqtt_disconnect(&client->mqttc); + if (1 != rc) { + // TODO: Disconnect failed + } + + if (-1 != client->mqttc.socketfd) { + close(client->mqttc.socketfd); + } + } + return ClientSuccess; +} + +client_error mqttc_client_destroy(mqttc_client_t *client) +{ + while (!neu_list_empty(&client->subscribe_list)) { + struct subscribe_tuple *tuple = neu_list_first(&client->subscribe_list); + neu_list_remove(&client->subscribe_list, tuple); + } + + if (NULL != &client->client_daemon) { + pthread_cancel(client->client_daemon); + } + + free(client); + return ClientSuccess; +} + +struct subscribe_tuple * +mqttc_client_subscribe_create(mqttc_client_t *client, const char *topic, + const int qos, const subscribe_handle handle) +{ + struct subscribe_tuple *tuple = + (struct subscribe_tuple *) malloc(sizeof(struct subscribe_tuple)); + + if (NULL == tuple) { + log_error("Failed to malloc client memory"); + return NULL; + } + + tuple->topic = strdup(topic); + tuple->qos = qos; + tuple->handle = handle; + tuple->subscribed = false; + tuple->client = client; + + return tuple; +} + +client_error mqttc_client_subscribe_destroy(struct subscribe_tuple *tuple) +{ + if (NULL != tuple) { + if (NULL != tuple->topic) { + free(tuple->topic); + } + free(tuple); + } + return ClientSuccess; +} + +client_error mqttc_client_subscribe_send(mqttc_client_t * client, + struct subscribe_tuple *tuple) +{ + int rc = mqtt_subscribe(&client->mqttc, tuple->topic, tuple->qos); + if (1 != rc) { + log_error("Failed to start subscribe, return code %d", rc); + return ClientSubscribeFailure; + } + return ClientSuccess; +} + +client_error mqttc_client_subscribe_add(mqttc_client_t * client, + struct subscribe_tuple *tuple) +{ + NEU_LIST_NODE_INIT(&tuple->node); + neu_list_append(&client->subscribe_list, tuple); + return ClientSuccess; +} + +client_error mqttc_client_subscribe_remove(mqttc_client_t *client, + const char * topic) +{ + struct subscribe_tuple *item; + struct subscribe_tuple *tuple; + NEU_LIST_FOREACH(&client->subscribe_list, item) + { + if (0 == strcmp(item->topic, topic)) { + tuple = item; + } + } + neu_list_remove(&client->subscribe_list, tuple); + return ClientSuccess; +} + +client_error mqttc_client_subscribe(mqttc_client_t *client, const char *topic, + const int qos, subscribe_handle handle) +{ + log_info("Subscribing to topic %s for using QoS%d", topic, qos); + struct subscribe_tuple *tuple = + mqttc_client_subscribe_create(client, topic, qos, handle); + if (NULL == tuple) { + return ClientSubscribeFailure; + } + + client_error error = mqttc_client_subscribe_add(client, tuple); + error = mqttc_client_subscribe_send(client, tuple); + if (ClientSuccess != error) { + return ClientSubscribeFailure; + } + return ClientSuccess; +} + +struct subscribe_tuple *mqttc_client_unsubscribe_create(mqttc_client_t *client, + const char * topic) +{ + struct subscribe_tuple *item; + NEU_LIST_FOREACH(&client->subscribe_list, item) + { + if (0 == strcmp(item->topic, topic)) { + return item; + } + } + + log_error("Cant find topic %s", topic); + return NULL; +} + +client_error mqttc_client_unsubscribe_send(mqttc_client_t * client, + struct subscribe_tuple *tuple) +{ + int rc = mqtt_unsubscribe(&client->mqttc, tuple->topic); + if (1 != rc) { + return ClientUnsubscribeFailure; + } + return ClientSuccess; +} + +client_error mqttc_client_unsubscribe(mqttc_client_t *client, const char *topic) +{ + if (NULL == client) { + return ClientIsNULL; + } + + struct subscribe_tuple *tuple = + mqttc_client_unsubscribe_create(client, topic); + if (NULL == tuple) { + return ClientUnsubscribeFailure; + } + + mqttc_client_unsubscribe_send(client, tuple); + + client_error error = mqttc_client_subscribe_remove(client, topic); + if (ClientSuccess != error) { + return error; + } + return ClientSuccess; +} + +client_error mqttc_client_publish(mqttc_client_t *client, const char *topic, + int qos, unsigned char *payload, size_t len) +{ + log_info("Publish to topic %s for using QoS%d", topic, qos); + int rc = mqtt_publish(&client->mqttc, topic, payload, len, qos); + if (1 != rc) { + log_error("Failed to start send message, return code %d", rc); + return ClientPublishFailure; + } + return ClientSuccess; +} + +client_error mqttc_client_close(mqttc_client_t *client) +{ + if (NULL != client) { + return ClientIsNULL; + } + + client_error rc = mqttc_client_disconnect(client); + rc = mqttc_client_destroy(client); + return rc; +} diff --git a/plugins/mqtt/mqttc_client.h b/plugins/mqtt/mqttc_client.h new file mode 100644 index 00000000..362e2765 --- /dev/null +++ b/plugins/mqtt/mqttc_client.h @@ -0,0 +1,49 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 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_PLUGIN_MQTTC_CLIENT +#define NEURON_PLUGIN_MQTTC_CLIENT + +#ifdef __cplusplus +extern "C" { +#endif + +#include +#include + +#include "mqtt_client.h" +#include "mqttc_client.h" + +typedef struct mqttc_client mqttc_client_t; + +client_error mqttc_client_open(option_t *option, void *context, + mqttc_client_t **p_client); +client_error mqttc_client_is_connected(mqttc_client_t *client); +client_error mqttc_client_subscribe(mqttc_client_t *client, const char *topic, + const int qos, subscribe_handle handle); +client_error mqttc_client_unsubscribe(mqttc_client_t *client, + const char * topic); +client_error mqttc_client_publish(mqttc_client_t *client, const char *topic, + int qos, unsigned char *payload, size_t len); +client_error mqttc_client_close(mqttc_client_t *client); + +#ifdef __cplusplus +} +#endif +#endif \ No newline at end of file diff --git a/plugins/mqtt/paho_client.c b/plugins/mqtt/paho_client.c index f78d9a25..ca406a42 100644 --- a/plugins/mqtt/paho_client.c +++ b/plugins/mqtt/paho_client.c @@ -24,7 +24,6 @@ #include #include -#include "option.h" #include "paho_client.h" #define UNUSED(x) (void) (x) diff --git a/plugins/mqtt/paho_client.h b/plugins/mqtt/paho_client.h index fc93e110..63dbb149 100644 --- a/plugins/mqtt/paho_client.h +++ b/plugins/mqtt/paho_client.h @@ -27,26 +27,11 @@ extern "C" { #include #include +#include "mqtt_client.h" + typedef struct paho_client paho_client_t; typedef struct subscribe_tuple subscribe_tuple_t; -typedef enum { - ClientSuccess = 0, - ClientIsNULL, - ClientConnectFailure, - ClientSubscribeTimeout, - ClientSubscribeListInitialFailure, - ClientSubscribeFailure, - ClientSubscribeAddListRepeat, - ClientSubscribeAddListFailure, - ClientUnsubscribeFailure, - ClientPublishFailure, -} client_error; - -typedef void (*subscribe_handle)(const char *topic_name, size_t topic_len, - void *payload, const size_t len, - void *context); - paho_client_t *paho_client_create(option_t *option, void *context); client_error paho_client_connect(paho_client_t *client); client_error paho_client_open(option_t *option, void *context, diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index 1525f624..3596c3df 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -29,15 +29,21 @@ #include "command_node.h" #include "command_rw.h" -#include "option.h" +#include "mqtt_client.h" +#include "mqttc_client.h" #include "paho_client.h" #define UNUSED(x) (void) (x) +#define MQTT_TYPE_PAHO 1; +#define MQTT_TYPE_MQTTC 2; +#define CONFIG_FILE "./neuron.yaml" +#define CONFIG_NODE "mqtt" + #define MQTT_SEND(client, topic, qos, json_str) \ { \ if (NULL != json_str) { \ - client_error error = paho_client_publish( \ + client_error error = mqttc_client_publish( \ client, topic, qos, (unsigned char *) json_str, \ strlen(json_str)); \ log_debug("Publish error code:%d, json:%s", error, json_str); \ @@ -58,10 +64,124 @@ struct neu_plugin { option_t option; paho_client_t * paho; neu_list context_list; - neu_parse_mqtt_t * parse_header; + int mqtt_client_type; + void * mqtt_client; }; -struct context *context_create() +static int mqtt_option_init(neu_plugin_t *plugin) +{ + char *clientid = neu_config_get_value(CONFIG_FILE, 2, CONFIG_NODE, "id"); + char *mqtt_version = + neu_config_get_value(CONFIG_FILE, 2, CONFIG_NODE, "mqtt_version"); + char *topic = neu_config_get_value(CONFIG_FILE, 2, CONFIG_NODE, "topic"); + char *qos = neu_config_get_value(CONFIG_FILE, 2, CONFIG_NODE, "qos"); + char *keepalive_interval = + neu_config_get_value(CONFIG_FILE, 2, CONFIG_NODE, "keepalive_interval"); + char *clean_session = + neu_config_get_value(CONFIG_FILE, 2, CONFIG_NODE, "clean_session"); + char *connection = neu_config_get_value(CONFIG_FILE, 3, CONFIG_NODE, + "broker", "connection"); + char *host = + neu_config_get_value(CONFIG_FILE, 3, CONFIG_NODE, "broker", "host"); + char *port = + neu_config_get_value(CONFIG_FILE, 3, CONFIG_NODE, "broker", "port"); + + // MQTT option + if (NULL == clientid) { + plugin->option.clientid = NULL; // Use random id + } else { + plugin->option.clientid = strdup(clientid); + free(clientid); + } + + if (NULL == mqtt_version) { + plugin->option.MQTT_version = 4; // Version 3.1.1 + } else { + plugin->option.MQTT_version = atoi(mqtt_version); + free(mqtt_version); + } + + if (NULL == topic) { + return -1; + } else { + plugin->option.topic = strdup(topic); + free(topic); + } + + if (NULL == qos) { + plugin->option.qos = 0; + } else { + plugin->option.qos = atoi(qos); + free(qos); + } + + if (NULL == connection) { + plugin->option.connection = strdup("tcp://"); + } else { + plugin->option.connection = strdup(connection); + free(connection); + } + + if (NULL == host) { + return -2; + } else { + plugin->option.host = strdup(host); + free(host); + } + + if (NULL == port) { + return -3; + } else { + plugin->option.port = strdup(port); + free(port); + } + + if (NULL == keepalive_interval) { + plugin->option.keepalive_interval = 20; + } else { + plugin->option.keepalive_interval = atoi(keepalive_interval); + free(keepalive_interval); + } + + if (NULL == clean_session) { + plugin->option.clean_session = 1; + } else { + plugin->option.clean_session = atoi(clean_session); + free(clean_session); + } + + return 0; +} + +static void mqtt_option_uninit(neu_plugin_t *plugin) +{ + if (NULL != plugin->option.clientid) { + free(plugin->option.clientid); + } + + if (NULL != plugin->option.topic) { + free(plugin->option.topic); + } + + if (NULL != plugin->option.connection) { + free(plugin->option.connection); + } + + if (NULL != plugin->option.host) { + free(plugin->option.host); + } + + if (NULL != plugin->option.port) { + free(plugin->option.port); + } +} + +static void context_list_init(neu_list *list) +{ + NEU_LIST_INIT(list, struct context, node); +} + +static struct context *context_create() { struct context *ctx = NULL; ctx = malloc(sizeof(struct context)); @@ -71,9 +191,14 @@ struct context *context_create() return ctx; } -void context_list_add(neu_list *list, struct context *ctx, int req_id, - neu_parse_mqtt_t *parse_header) +static void context_list_add(neu_list *list, int req_id, + neu_parse_mqtt_t *parse_header) { + struct context *ctx = context_create(); + if (NULL == ctx) { + return; + } + ctx->req_id = req_id; ctx->parse_header.function = parse_header->function; ctx->parse_header.uuid = strdup(parse_header->uuid); @@ -82,7 +207,7 @@ void context_list_add(neu_list *list, struct context *ctx, int req_id, neu_list_append(list, ctx); } -struct context *context_list_find(neu_list *list, const int id) +static struct context *context_list_find(neu_list *list, const int id) { struct context *item; NEU_LIST_FOREACH(list, item) @@ -94,7 +219,7 @@ struct context *context_list_find(neu_list *list, const int id) return NULL; } -void context_destroy(struct context *ctx) +static void context_destroy(struct context *ctx) { if (NULL != ctx) { if (NULL != ctx->parse_header.uuid) { @@ -104,13 +229,12 @@ void context_destroy(struct context *ctx) } } -void context_list_remove(neu_list *list, struct context *ctx) +static void context_list_remove(neu_list *list, struct context *ctx) { - UNUSED(list); - UNUSED(ctx); + neu_list_remove(list, ctx); } -void context_list_destroy(neu_list *list) +static void context_list_destroy(neu_list *list) { while (!neu_list_empty(list)) { struct context *ctx = neu_list_first(list); @@ -119,22 +243,12 @@ void context_list_destroy(neu_list *list) } } -static int plugin_subscribe(neu_plugin_t *plugin, const char *topic, - const int qos, subscribe_handle handle) -{ - client_error error = - paho_client_subscribe(plugin->paho, topic, qos, handle); - if (ClientSubscribeFailure == error || - ClientSubscribeAddListFailure == error) { - neu_panic("Subscribe Failure"); - } - return 0; -} - static void plugin_response_handle(const char *topic_name, size_t topic_len, void *payload, const size_t len, void *context) { + UNUSED(topic_len); + neu_plugin_t *plugin = (neu_plugin_t *) context; if (NULL == topic_name || NULL == payload) { @@ -158,8 +272,6 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, return; } - plugin->parse_header = mqtt; - switch (mqtt->function) { case NEU_MQTT_OP_GET_GROUP_CONFIG: { neu_parse_get_group_config_req_t *req = NULL; @@ -202,7 +314,9 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, rc = neu_parse_decode_read(json_str, &req); if (0 == rc) { int req_id = command_read_once_request(plugin, mqtt, req); - UNUSED(req_id); + if (0 < req_id) { + context_list_add(&plugin->context_list, req_id, mqtt); + } neu_parse_decode_read_free(req); } break; @@ -212,7 +326,9 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, rc = neu_parse_decode_write(json_str, &req); if (0 == rc) { int req_id = command_write_request(plugin, mqtt, req); - UNUSED(req_id); + if (0 < req_id) { + context_list_add(&plugin->context_list, req_id, mqtt); + } neu_parse_decode_write_free(req); } break; @@ -293,20 +409,18 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, break; } - MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + MQTT_SEND(plugin->mqtt_client, "neuronlite/response", 0, ret_str); - // if (NULL != mqtt) { - // if (NULL != mqtt->uuid) { - // free(mqtt->uuid); - // } - // free(mqtt); - // } + if (NULL != mqtt) { + if (NULL != mqtt->uuid) { + free(mqtt->uuid); + } + free(mqtt); + } if (NULL != json_str) { free(json_str); } - - UNUSED(topic_len); } static neu_plugin_t *mqtt_plugin_open(neu_adapter_t * adapter, @@ -340,57 +454,34 @@ static int mqtt_plugin_close(neu_plugin_t *plugin) static int mqtt_plugin_init(neu_plugin_t *plugin) { - const char *clientid = - neu_config_get_value("./neuron.yaml", 2, "mqtt", "id"); - const char *mqtt_version = - neu_config_get_value("./neuron.yaml", 2, "mqtt", "mqtt_version"); - const char *topic = - neu_config_get_value("./neuron.yaml", 2, "mqtt", "topic"); - const char *qos = neu_config_get_value("./neuron.yaml", 2, "mqtt", "qos"); - const char *keepalive_interval = - neu_config_get_value("./neuron.yaml", 2, "mqtt", "keepalive_interval"); - const char *clean_session = - neu_config_get_value("./neuron.yaml", 2, "mqtt", "clean_session"); - const char *connection = neu_config_get_value("./neuron.yaml", 3, "mqtt", - "broker", "connection"); - const char *host = - neu_config_get_value("./neuron.yaml", 3, "mqtt", "broker", "host"); - const char *port = - neu_config_get_value("./neuron.yaml", 3, "mqtt", "broker", "port"); + // Context list init + context_list_init(&plugin->context_list); - // MQTT option - plugin->option.clientid = strdup(clientid); - plugin->option.MQTT_version = atoi(mqtt_version); - plugin->option.topic = strdup(topic); - plugin->option.qos = atoi(qos); - plugin->option.connection = strdup(connection); - plugin->option.host = strdup(host); - plugin->option.port = strdup(port); - plugin->option.keepalive_interval = atoi(keepalive_interval); - plugin->option.clean_session = atoi(clean_session); - - plugin->parse_header = NULL; - - NEU_LIST_INIT(&plugin->context_list, struct context, node); - - // Paho mqtt client setup - client_error error = - paho_client_open(&plugin->option, plugin, &plugin->paho); - if (ClientIsNULL == error) { + int rc = mqtt_option_init(plugin); + if (0 != rc) { + log_error("mqtt option init fail:%d", rc); return -1; } - plugin_subscribe(plugin, "neuronlite/request", 0, plugin_response_handle); - log_info("Initialize plugin: %s", neu_plugin_module.module_name); + // MQTT-C client setup + client_error error = mqttc_client_open( + &plugin->option, plugin, (mqttc_client_t **) &plugin->mqtt_client); + log_info("mqttc open:%d", error); + error = mqttc_client_subscribe(plugin->mqtt_client, "neuronlite/request", 0, + plugin_response_handle); + log_info("mqttc subscribe:%d", error); + + log_info("Initialize plugin: %s", neu_plugin_module.module_name); return 0; } static int mqtt_plugin_uninit(neu_plugin_t *plugin) { - paho_client_close(plugin->paho); - log_info("Uninitialize plugin: %s", neu_plugin_module.module_name); + mqttc_client_close(plugin->mqtt_client); + mqtt_option_uninit(plugin); + context_list_destroy(&plugin->context_list); return 0; } @@ -418,18 +509,30 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) switch (req->req_type) { case NEU_REQRESP_READ_RESP: { neu_reqresp_read_resp_t *read_resp; - read_resp = (neu_reqresp_read_resp_t *) req->buf; - char *json_str = command_read_once_response( - plugin, plugin->parse_header, read_resp->data_val); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str); + read_resp = (neu_reqresp_read_resp_t *) req->buf; + + struct context *ctx = NULL; + ctx = context_list_find(&plugin->context_list, req->req_id); + if (NULL != ctx) { + char *json_str = command_read_once_response( + plugin, &ctx->parse_header, read_resp->data_val); + MQTT_SEND(plugin->mqtt_client, "neuronlite/response", 0, json_str); + context_list_remove(&plugin->context_list, ctx); + } break; } case NEU_REQRESP_WRITE_RESP: { neu_reqresp_write_resp_t *write_resp; write_resp = (neu_reqresp_write_resp_t *) req->buf; - char *json_str = - command_write_response(plugin, "", write_resp->data_val); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str); + + struct context *ctx = NULL; + ctx = context_list_find(&plugin->context_list, req->req_id); + if (NULL != ctx) { + char *json_str = command_write_response(plugin, &ctx->parse_header, + write_resp->data_val); + MQTT_SEND(plugin->mqtt_client, "neuronlite/response", 0, json_str); + context_list_remove(&plugin->context_list, ctx); + } break; } case NEU_REQRESP_TRANS_DATA: { @@ -437,7 +540,7 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) neu_data = (neu_reqresp_data_t *) req->buf; char *json_str = command_read_cycle_response(plugin, neu_data->data_val); - MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str); + MQTT_SEND(plugin->mqtt_client, "neuronlite/response", 0, json_str); break; } case NEU_REQRESP_WRITE_DATA: { diff --git a/plugins/mqtt/posix_sockets.h b/plugins/mqtt/posix_sockets.h new file mode 100644 index 00000000..adc4532b --- /dev/null +++ b/plugins/mqtt/posix_sockets.h @@ -0,0 +1,80 @@ +#if !defined(__POSIX_SOCKET_TEMPLATE_H__) +#define __POSIX_SOCKET_TEMPLATE_H__ + +#include +#include +#if !defined(WIN32) +#include +#include +#endif +#if defined(__VMS) +#include +#endif +#include + +/* + A template for opening a non-blocking POSIX socket. +*/ +int open_nb_socket(const char *addr, const char *port) +{ + struct addrinfo hints = { 0 }; + + hints.ai_family = AF_UNSPEC; /* IPv4 or IPv6 */ + hints.ai_socktype = SOCK_STREAM; /* Must be TCP */ + int sockfd = -1; + int rv; + struct addrinfo *p, *servinfo; + + /* get address information */ + rv = getaddrinfo(addr, port, &hints, &servinfo); + if (rv != 0) { + fprintf(stderr, "Failed to open socket (getaddrinfo): %s\n", + gai_strerror(rv)); + return -1; + } + + /* open the first possible socket */ + for (p = servinfo; p != NULL; p = p->ai_next) { + sockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol); + if (sockfd == -1) + continue; + + /* connect to server */ + rv = connect(sockfd, p->ai_addr, p->ai_addrlen); + if (rv == -1) { + close(sockfd); + sockfd = -1; + continue; + } + break; + } + + /* free servinfo */ + freeaddrinfo(servinfo); + + /* make non-blocking */ +#if !defined(WIN32) + if (sockfd != -1) + fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFL) | O_NONBLOCK); +#else + if (sockfd != INVALID_SOCKET) { + int iMode = 1; + ioctlsocket(sockfd, FIONBIO, &iMode); + } +#endif +#if defined(__VMS) + /* + OpenVMS only partially implements fcntl. It works on file descriptors + but silently fails on socket descriptors. So we need to fall back on + to the older ioctl system to set non-blocking IO + */ + int on = 1; + if (sockfd != -1) + ioctl(sockfd, FIONBIO, &on); +#endif + + /* return the new socket fd */ + return sockfd; +} + +#endif