Add the implementation of mqtt-c client

- Add README
- Add context list to store request response context

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

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

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})

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;
 }

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
 }

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

plugins/mqtt/mqttc_client.c

@@ -0,0 +1,396 @@
+
+/**
+ * @file
+ * A simple subscriber program that performs automatic reconnections.
+ */
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#include <mqtt.h>
+#include <neuron.h>
+
+#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;
+}

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 <stddef.h>
+#include <stdint.h>
+
+#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

plugins/mqtt/paho_client.c

@@ -24,7 +24,6 @@
 #include <MQTTAsync.h>
 #include <neuron.h>
 
-#include "option.h"
 #include "paho_client.h"
 
 #define UNUSED(x) (void) (x)

plugins/mqtt/paho_client.h

@@ -27,26 +27,11 @@ extern "C" {
 #include <stddef.h>
 #include <stdint.h>
 
+#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,

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: {

plugins/mqtt/posix_sockets.h

@@ -0,0 +1,80 @@
+#if !defined(__POSIX_SOCKET_TEMPLATE_H__)
+#define __POSIX_SOCKET_TEMPLATE_H__
+
+#include <stdio.h>
+#include <sys/types.h>
+#if !defined(WIN32)
+#include <netdb.h>
+#include <sys/socket.h>
+#endif
+#if defined(__VMS)
+#include <ioctl.h>
+#endif
+#include <fcntl.h>
+
+/*
+    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