Merge pull request #1263 from fengzeroz/v2.4

version: 2.4.6

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.4.4
-  UI_VERSION: 2.4.3
+  VERSION: 2.4.6
+  UI_VERSION: 2.4.4
   BRANCH: v2.4
 
 jobs:

.gitignore

@@ -16,4 +16,7 @@ tmp
 *.deb
 *.rpm
 neuron-sdk
-.vscode
+.vscode
+log.html
+output.xml
+report.html

cmake/cross.cmake

@@ -17,3 +17,7 @@ set(CMAKE_FIND_ROOT_PATH_MODE_PROGRAM NEVER)
 set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
 set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
+
+if(CMAKE_SYSTEM_PROCESSOR MATCHES "mips32")
+	link_libraries(atomic)
+endif()

ft/resource/error.resource

@@ -71,4 +71,7 @@ ${NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED} 	       	3008
 ${NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE}	3009
 ${NEU_ERR_PLUGIN_NOT_RUNNING}			3010
 ${NEU_ERR_PLUGIN_TAG_NOT_READY}           	3011
-${NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER}     	3012
+${NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER}     	3012
+${NEU_ERR_PLUGIN_NAME_TOO_LONG}         	3013
+${NEU_ERR_PLUGIN_NOT_FOUND}             	3014
+${NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE}     	3015

include/neuron/connection/mqtt_client.h

@@ -114,6 +114,8 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
                               const char *topic, void *data,
                               neu_mqtt_client_subscribe_cb_t cb);
 
+int neu_mqtt_client_unsubscribe(neu_mqtt_client_t *client, const char *topic);
+
 #ifdef __cplusplus
 }
 #endif

include/neuron/errcodes.h

@@ -113,6 +113,7 @@ typedef enum {
     NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER     = 3012,
     NEU_ERR_PLUGIN_NAME_TOO_LONG           = 3013,
     NEU_ERR_PLUGIN_NOT_FOUND               = 3014,
+    NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE     = 3015,
 
     NEU_ERR_MQTT_FAILURE                        = 4000,
     NEU_ERR_MQTT_NO_CERTFILESET                 = 4001,

include/neuron/utils/protocol_buf.h

@@ -192,4 +192,16 @@ neu_protocol_unpack_buf_used_size(neu_protocol_unpack_buf_t *buf)
     return buf->offset;
 }
 
+inline static void
+neu_protocol_unpack_buf_use_all(neu_protocol_unpack_buf_t *buf)
+{
+    buf->offset = buf->size;
+}
+
+inline static uint16_t
+neu_protocol_unpack_buf_size(neu_protocol_unpack_buf_t *buf)
+{
+    return buf->size;
+}
+
 #endif

include/neuron/utils/time.h

@@ -26,6 +26,7 @@ extern "C" {
 #include <stdint.h>
 #include <stdlib.h>
 #include <sys/time.h>
+#include <time.h>
 
 static inline int64_t neu_time_ms()
 {
@@ -34,6 +35,15 @@ static inline int64_t neu_time_ms()
     return (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000;
 }
 
+static inline void neu_msleep(unsigned msec)
+{
+    struct timespec tv = {
+        .tv_sec  = msec / 1000,
+        .tv_nsec = (msec % 1000) * 1000000,
+    };
+    nanosleep(&tv, NULL);
+}
+
 #ifdef __cplusplus
 }
 #endif

persistence/0033_2.4.5_mqtt_upload_topic.sql

@@ -0,0 +1,59 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2023 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.
+ **/
+BEGIN TRANSACTION;
+
+-- for backward compatibility with version 2.3,
+-- migrate MQTT plugin upload-topic to subscription parameters.
+WITH
+  topics AS (
+    SELECT
+      name AS app,
+      json_extract(json(setting), '$.params.upload-topic') AS topic
+    FROM
+      nodes
+      JOIN settings
+    WHERE
+      name = node_name
+      AND plugin_name = 'MQTT'
+  )
+UPDATE
+  subscriptions
+SET
+  params = (
+    SELECT
+      json_object('params', json_object('topic', topic)) AS params
+    FROM
+      topics
+    WHERE
+      topics.app = subscriptions.app_name
+      AND topics.topic NOT NULL
+  )
+WHERE
+  EXISTS (
+    SELECT
+      topic
+    FROM
+      topics
+    WHERE
+      topics.app = subscriptions.app_name
+      AND topics.topic NOT NULL
+      AND subscriptions.params IS NULL
+  );
+
+COMMIT;

plugins/file/file_plugin.c

@@ -81,8 +81,7 @@ static int driver_close(neu_plugin_t *plugin)
 
 static int driver_init(neu_plugin_t *plugin)
 {
-    plugin->events            = neu_event_new();
-    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
+    plugin->events = neu_event_new();
 
     return 0;
 }
@@ -97,14 +96,14 @@ static int driver_uninit(neu_plugin_t *plugin)
 
 static int driver_start(neu_plugin_t *plugin)
 {
-    (void) plugin;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
 
     return 0;
 }
 
 static int driver_stop(neu_plugin_t *plugin)
 {
-    (void) plugin;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
 
     return 0;
 }
@@ -121,7 +120,8 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
         return -1;
     }
 
-    plugin->file_length = file_length.v.val_int;
+    plugin->file_length       = file_length.v.val_int;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
 
     return 0;
 }

plugins/mqtt/mqtt.json

@@ -57,6 +57,30 @@
       ]
     }
   },
+  "write-req-topic": {
+    "name": "Write Request Topic",
+    "name_zh": "写请求主题",
+    "description": "MQTT topic to receive write request messages.",
+    "description_zh": "接收点位写入请求的 MQTT 主题。",
+    "attribute": "required",
+    "type": "string",
+    "default": "/neuron/${node-name}/write/req",
+    "valid": {
+      "length": 255
+    }
+  },
+  "write-resp-topic": {
+    "name": "Write Response Topic",
+    "name_zh": "写响应主题",
+    "description": "MQTT topic to send write response messages.",
+    "description_zh": "发送点位写入响应的 MQTT 主题。",
+    "attribute": "required",
+    "type": "string",
+    "default": "/neuron/${node-name}/write/resp",
+    "valid": {
+      "length": 255
+    }
+  },
   "offline-cache": {
     "name": "Offline Data Caching",
     "name_zh": "离线缓存",
@@ -218,4 +242,4 @@
       "length": 256
     }
   }
-}
+}

plugins/mqtt/mqtt_config.c

@@ -17,6 +17,7 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include "utils/asprintf.h"
 #include "json/json.h"
 #include "json/neu_json_param.h"
 
@@ -203,6 +204,18 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
     };
     neu_json_elem_t format          = { .name = "format", .t = NEU_JSON_INT };
+    neu_json_elem_t write_req_topic = {
+        .name      = "write-req-topic",
+        .t         = NEU_JSON_STR,
+        .v.val_str = NULL,
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
+    neu_json_elem_t write_resp_topic = {
+        .name      = "write-resp-topic",
+        .t         = NEU_JSON_STR,
+        .v.val_str = NULL,
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
     neu_json_elem_t offline_cache   = { .name = "offline-cache",
                                       .t    = NEU_JSON_BOOL };
     neu_json_elem_t cache_mem_size  = { .name = "cache-mem-size",
@@ -224,8 +237,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         return -1;
     }
 
-    ret = neu_parse_param(setting, &err_param, 5, &client_id, &format, &qos,
-                          &host, &port);
+    ret = neu_parse_param(setting, &err_param, 7, &client_id, &qos, &format,
+                          &write_req_topic, &write_resp_topic, &host, &port);
     if (0 != ret) {
         plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
         goto error;
@@ -251,6 +264,22 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
+    // write request topic
+    if (NULL == write_req_topic.v.val_str &&
+        0 > neu_asprintf(&write_req_topic.v.val_str, "/neuron/%s/write/req",
+                         plugin->common.name)) {
+        plog_error(plugin, "setting write request topic error");
+        goto error;
+    }
+
+    // write response topic
+    if (NULL == write_resp_topic.v.val_str &&
+        0 > neu_asprintf(&write_resp_topic.v.val_str, "/neuron/%s/write/resp",
+                         plugin->common.name)) {
+        plog_error(plugin, "setting write response topic error");
+        goto error;
+    }
+
     // offline cache
     ret = parse_cache_params(plugin, setting, &offline_cache, &cache_mem_size,
                              &cache_disk_size);
@@ -287,55 +316,61 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
-    config->client_id       = client_id.v.val_str;
-    config->qos             = qos.v.val_int;
-    config->format          = format.v.val_int;
-    config->cache           = offline_cache.v.val_bool;
-    config->cache_mem_size  = cache_mem_size.v.val_int * MB;
-    config->cache_disk_size = cache_disk_size.v.val_int * MB;
-    config->host            = host.v.val_str;
-    config->port            = port.v.val_int;
-    config->username        = username.v.val_str;
-    config->password        = password.v.val_str;
-    config->ca              = ca.v.val_str;
-    config->cert            = cert.v.val_str;
-    config->key             = key.v.val_str;
-    config->keypass         = keypass.v.val_str;
+    config->client_id        = client_id.v.val_str;
+    config->qos              = qos.v.val_int;
+    config->format           = format.v.val_int;
+    config->write_req_topic  = write_req_topic.v.val_str;
+    config->write_resp_topic = write_resp_topic.v.val_str;
+    config->cache            = offline_cache.v.val_bool;
+    config->cache_mem_size   = cache_mem_size.v.val_int * MB;
+    config->cache_disk_size  = cache_disk_size.v.val_int * MB;
+    config->host             = host.v.val_str;
+    config->port             = port.v.val_int;
+    config->username         = username.v.val_str;
+    config->password         = password.v.val_str;
+    config->ca               = ca.v.val_str;
+    config->cert             = cert.v.val_str;
+    config->key              = key.v.val_str;
+    config->keypass          = keypass.v.val_str;
 
-    plog_info(plugin, "config client-id      : %s", config->client_id);
-    plog_info(plugin, "config qos            : %d", config->qos);
-    plog_info(plugin, "config format         : %s",
+    plog_info(plugin, "config client-id       : %s", config->client_id);
+    plog_info(plugin, "config qos             : %d", config->qos);
+    plog_info(plugin, "config format          : %s",
               mqtt_upload_format_str(config->format));
-    plog_info(plugin, "config cache          : %zu", config->cache);
-    plog_info(plugin, "config cache-mem-size : %zu", config->cache_mem_size);
-    plog_info(plugin, "config cache-disk-size: %zu", config->cache_disk_size);
-    plog_info(plugin, "config host           : %s", config->host);
-    plog_info(plugin, "config port           : %" PRIu16, config->port);
+    plog_info(plugin, "config write-req-topic : %s", config->write_req_topic);
+    plog_info(plugin, "config write-resp-topic: %s", config->write_resp_topic);
+    plog_info(plugin, "config cache           : %zu", config->cache);
+    plog_info(plugin, "config cache-mem-size  : %zu", config->cache_mem_size);
+    plog_info(plugin, "config cache-disk-size : %zu", config->cache_disk_size);
+    plog_info(plugin, "config host            : %s", config->host);
+    plog_info(plugin, "config port            : %" PRIu16, config->port);
 
     if (config->username) {
-        plog_info(plugin, "config username       : %s", config->username);
+        plog_info(plugin, "config username        : %s", config->username);
     }
     if (config->password) {
-        plog_info(plugin, "config password       : %s",
+        plog_info(plugin, "config password        : %s",
                   0 == strlen(config->password) ? "" : placeholder);
     }
     if (config->ca) {
-        plog_info(plugin, "config ca             : %s", placeholder);
+        plog_info(plugin, "config ca              : %s", placeholder);
     }
     if (config->cert) {
-        plog_info(plugin, "config cert           : %s", placeholder);
+        plog_info(plugin, "config cert            : %s", placeholder);
     }
     if (config->key) {
-        plog_info(plugin, "config key            : %s", placeholder);
+        plog_info(plugin, "config key             : %s", placeholder);
     }
     if (config->keypass) {
-        plog_info(plugin, "config keypass        : %s", placeholder);
+        plog_info(plugin, "config keypass         : %s", placeholder);
     }
 
     return 0;
 
 error:
     free(err_param);
+    free(write_req_topic.v.val_str);
+    free(write_resp_topic.v.val_str);
     free(host.v.val_str);
     free(username.v.val_str);
     free(password.v.val_str);
@@ -349,6 +384,8 @@ error:
 void mqtt_config_fini(mqtt_config_t *config)
 {
     free(config->client_id);
+    free(config->write_req_topic);
+    free(config->write_resp_topic);
     free(config->host);
     free(config->username);
     free(config->password);

plugins/mqtt/mqtt_config.h

@@ -48,20 +48,22 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
 }
 
 typedef struct {
-    char *               client_id;       // client id
-    mqtt_upload_format_e format;          // upload format
-    neu_mqtt_qos_e       qos;             // message QoS
-    size_t               cache;           // cache enable flag
-    size_t               cache_mem_size;  // cache memory size in bytes
-    size_t               cache_disk_size; // cache disk size in bytes
-    char *               host;            // broker host
-    uint16_t             port;            // broker port
-    char *               username;        // user name
-    char *               password;        // user password
-    char *               ca;              // CA
-    char *               cert;            // client cert
-    char *               key;             // client key
-    char *               keypass;         // client key password
+    char *               client_id;        // client id
+    neu_mqtt_qos_e       qos;              // message QoS
+    mqtt_upload_format_e format;           // upload format
+    char *               write_req_topic;  // write request topic
+    char *               write_resp_topic; // write response topic
+    size_t               cache;            // cache enable flag
+    size_t               cache_mem_size;   // cache memory size in bytes
+    size_t               cache_disk_size;  // cache disk size in bytes
+    char *               host;             // broker host
+    uint16_t             port;             // broker port
+    char *               username;         // user name
+    char *               password;         // user password
+    char *               ca;               // CA
+    char *               cert;             // client cert
+    char *               key;              // client key
+    char *               keypass;          // client key password
 } mqtt_config_t;
 
 int  mqtt_config_parse(neu_plugin_t *plugin, const char *setting,

plugins/mqtt/mqtt_handle.c

@@ -373,7 +373,7 @@ int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
         goto end;
     }
 
-    char *         topic = plugin->write_resp_topic;
+    char *         topic = plugin->config.write_resp_topic;
     neu_mqtt_qos_e qos   = plugin->config.qos;
     rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
     json_str = NULL;

plugins/mqtt/mqtt_plugin.c

@@ -18,6 +18,7 @@
  **/
 
 #include "utils/asprintf.h"
+#include "utils/time.h"
 
 #include "mqtt_config.h"
 #include "mqtt_handle.h"
@@ -81,10 +82,6 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin)
     plugin->read_req_topic = NULL;
     free(plugin->read_resp_topic);
     plugin->read_resp_topic = NULL;
-    free(plugin->write_req_topic);
-    plugin->write_req_topic = NULL;
-    free(plugin->write_resp_topic);
-    plugin->write_resp_topic = NULL;
 
     route_tbl_free(plugin->route_tbl);
 
@@ -159,6 +156,65 @@ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
     return rv;
 }
 
+static int create_topic(neu_plugin_t *plugin)
+{
+    if (plugin->read_req_topic) {
+        // topics already created
+        return 0;
+    }
+
+    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
+                 plugin->common.name);
+    if (NULL == plugin->read_req_topic) {
+        return -1;
+    }
+
+    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
+                 plugin->common.name);
+    if (NULL == plugin->read_resp_topic) {
+        free(plugin->read_req_topic);
+        plugin->read_req_topic = NULL;
+        return -1;
+    }
+
+    return 0;
+}
+
+static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
+{
+    if (0 != create_topic(plugin)) {
+        plog_error(plugin, "create topics fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, config->qos,
+                                  plugin->read_req_topic, plugin,
+                                  handle_read_req)) {
+        plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
+        return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, config->qos,
+                                  config->write_req_topic, plugin,
+                                  handle_write_req)) {
+        plog_error(plugin, "subscribe [%s] fail",
+                   plugin->config.write_req_topic);
+        return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+    }
+
+    return 0;
+}
+
+static int unsubscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
+{
+    neu_mqtt_client_unsubscribe(plugin->client, plugin->read_req_topic);
+    neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic);
+    neu_msleep(100); // wait for message completion
+    return 0;
+}
+
 static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
 {
     int           rv          = 0;
@@ -181,7 +237,8 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
         }
     } else if (neu_mqtt_client_is_open(plugin->client)) {
         started = true;
-        rv      = neu_mqtt_client_close(plugin->client);
+        unsubscribe(plugin, &plugin->config);
+        rv = neu_mqtt_client_close(plugin->client);
         if (0 != rv) {
             plog_error(plugin, "neu_mqtt_client_close fail");
             rv = NEU_ERR_EINTERNAL;
@@ -195,10 +252,15 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
         goto error;
     }
 
-    if (started && 0 != neu_mqtt_client_open(plugin->client)) {
-        plog_error(plugin, "neu_mqtt_client_open fail");
-        rv = NEU_ERR_MQTT_CONNECT_FAILURE;
-        goto error;
+    if (started) {
+        if (0 != neu_mqtt_client_open(plugin->client)) {
+            plog_error(plugin, "neu_mqtt_client_open fail");
+            rv = NEU_ERR_MQTT_CONNECT_FAILURE;
+            goto error;
+        }
+        if (0 != (rv = subscribe(plugin, &config))) {
+            goto error;
+        }
     }
 
     if (plugin->config.host) {
@@ -216,52 +278,6 @@ error:
     return rv;
 }
 
-static int create_topic(neu_plugin_t *plugin)
-{
-    if (plugin->read_req_topic) {
-        // topics already created
-        return 0;
-    }
-
-    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
-                 plugin->common.name);
-    if (NULL == plugin->read_req_topic) {
-        return -1;
-    }
-
-    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
-                 plugin->common.name);
-    if (NULL == plugin->read_resp_topic) {
-        free(plugin->read_req_topic);
-        plugin->read_req_topic = NULL;
-        return -1;
-    }
-
-    neu_asprintf(&plugin->write_req_topic, "/neuron/%s/write/req",
-                 plugin->common.name);
-    if (NULL == plugin->write_req_topic) {
-        free(plugin->read_req_topic);
-        plugin->read_req_topic = NULL;
-        free(plugin->read_resp_topic);
-        plugin->read_resp_topic = NULL;
-        return -1;
-    }
-
-    neu_asprintf(&plugin->write_resp_topic, "/neuron/%s/write/resp",
-                 plugin->common.name);
-    if (NULL == plugin->write_resp_topic) {
-        free(plugin->read_req_topic);
-        plugin->read_req_topic = NULL;
-        free(plugin->read_resp_topic);
-        plugin->read_resp_topic = NULL;
-        free(plugin->write_req_topic);
-        plugin->write_req_topic = NULL;
-        return -1;
-    }
-
-    return 0;
-}
-
 static int mqtt_plugin_start(neu_plugin_t *plugin)
 {
     int         rv          = 0;
@@ -279,29 +295,7 @@ static int mqtt_plugin_start(neu_plugin_t *plugin)
         goto end;
     }
 
-    if (0 != create_topic(plugin)) {
-        plog_error(plugin, "create topics fail");
-        rv = NEU_ERR_EINTERNAL;
-        goto end;
-    }
-
-    if (0 !=
-        neu_mqtt_client_subscribe(plugin->client, plugin->config.qos,
-                                  plugin->read_req_topic, plugin,
-                                  handle_read_req)) {
-        plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
-        rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
-        goto end;
-    }
-
-    if (0 !=
-        neu_mqtt_client_subscribe(plugin->client, plugin->config.qos,
-                                  plugin->write_req_topic, plugin,
-                                  handle_write_req)) {
-        plog_error(plugin, "subscribe [%s] fail", plugin->write_req_topic);
-        rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
-        goto end;
-    }
+    rv = subscribe(plugin, &plugin->config);
 
 end:
     if (0 == rv) {
@@ -317,6 +311,7 @@ end:
 static int mqtt_plugin_stop(neu_plugin_t *plugin)
 {
     if (plugin->client) {
+        unsubscribe(plugin, &plugin->config);
         neu_mqtt_client_close(plugin->client);
         plog_info(plugin, "mqtt client closed");
     }

plugins/mqtt/mqtt_plugin.h

@@ -49,8 +49,6 @@ struct neu_plugin {
     int64_t             cache_metric_update_ts;
     char *              read_req_topic;
     char *              read_resp_topic;
-    char *              write_req_topic;
-    char *              write_resp_topic;
     route_entry_t *     route_tbl;
 };
 

src/adapter/adapter.c

@@ -124,8 +124,8 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
 
     rv = nng_pair1_open(&adapter->sock);
     assert(rv == 0);
-    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 128);
-    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 128);
+    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 8192);
     nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd);
 

src/adapter/driver/driver.c

@@ -19,6 +19,7 @@
 #include <assert.h>
 #include <math.h>
 #include <netinet/in.h>
+#include <pthread.h>
 #include <stdlib.h>
 
 #include <nng/nng.h>
@@ -37,12 +38,20 @@
 #include "errcodes.h"
 #include "tag.h"
 
+typedef struct to_be_write_tag {
+    neu_datatag_t *tag;
+    neu_value_u    value;
+    void *         req;
+} to_be_write_tag_t;
+
 typedef struct group {
     char *name;
 
-    int64_t      timestamp;
-    neu_group_t *group;
-    UT_array *   static_tags;
+    int64_t         timestamp;
+    neu_group_t *   group;
+    UT_array *      static_tags;
+    UT_array *      wt_tags;
+    pthread_mutex_t wt_mtx;
 
     neu_event_timer_t *report;
     neu_event_timer_t *read;
@@ -74,6 +83,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
                    neu_dvalue_t value);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
 static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
+static void     store_write_tag(group_t *group, to_be_write_tag_t *tag);
 
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 {
@@ -182,7 +192,13 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
         free(el->name);
         utarray_free(el->grp.tags);
 
+        utarray_foreach(el->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(el->static_tags);
+        utarray_free(el->wt_tags);
         neu_group_destroy(el->group);
         free(el);
     }
@@ -414,8 +430,12 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             driver->adapter.cb_funs.response(&driver->adapter, req, &error);
             free(req);
         } else {
-            driver->adapter.module->intf_funs->driver.write_tag(
-                driver->adapter.plugin, (void *) req, tag, cmd->value.value);
+            to_be_write_tag_t wtag = { 0 };
+            wtag.req               = (void *) req;
+            wtag.value             = cmd->value.value;
+            wtag.tag               = neu_tag_dup(tag);
+
+            store_write_tag(g, &wtag);
         }
 
         neu_tag_free(tag);
@@ -429,6 +449,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
 int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
                                  uint32_t interval)
 {
+    UT_icd   icd  = { sizeof(to_be_write_tag_t), NULL, NULL, NULL };
     group_t *find = NULL;
     int      ret  = NEU_ERR_GROUP_EXIST;
 
@@ -443,6 +464,10 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
             .type        = NEU_EVENT_TIMER_NOBLOCK,
         };
 
+        pthread_mutex_init(&find->wt_mtx, NULL);
+
+        utarray_new(find->wt_tags, &icd);
+
         find->driver         = driver;
         find->name           = strdup(name);
         find->group          = neu_group_new(name, interval);
@@ -530,11 +555,18 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
             neu_driver_cache_del(driver->cache, name, tag->name);
         }
 
+        utarray_foreach(find->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(find->static_tags);
         utarray_free(find->grp.tags);
+        utarray_free(find->wt_tags);
         neu_group_destroy(find->group);
         free(find);
 
+        pthread_mutex_destroy(&find->wt_mtx);
         neu_adapter_del_group_metrics(&driver->adapter, name);
         ret = NEU_ERR_SUCCESS;
     }
@@ -892,6 +924,17 @@ static int read_callback(void *usr_data)
         return 0;
     }
 
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_foreach(group->wt_tags, to_be_write_tag_t *, wtag)
+    {
+        group->driver->adapter.module->intf_funs->driver.write_tag(
+            group->driver->adapter.plugin, (void *) wtag->req, wtag->tag,
+            wtag->value);
+        neu_tag_free(wtag->tag);
+    }
+    utarray_clear(group->wt_tags);
+    pthread_mutex_unlock(&group->wt_mtx);
+
     if (neu_group_is_change(group->group, group->timestamp)) {
         neu_group_change_test(group->group, group->timestamp, (void *) group,
                               group_change);
@@ -1199,3 +1242,10 @@ static void read_group(int64_t timestamp, int64_t timeout,
         }
     }
 }
+
+static void store_write_tag(group_t *group, to_be_write_tag_t *tag)
+{
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_push_back(group->wt_tags, tag);
+    pthread_mutex_unlock(&group->wt_mtx);
+}

src/connection/connection.c

@@ -33,6 +33,10 @@
 
 #include "connection/neu_connection.h"
 
+#ifndef CMSPAR
+#define CMSPAR 010000000000 /* mark or space (stick) parity */
+#endif
+
 struct tcp_client {
     int                fd;
     struct sockaddr_in client;
@@ -397,7 +401,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                    conn->fd, len, ret, strerror(errno), errno);
     }
 
-    if (errno == EPIPE || ret == -1 || ret == 0) {
+    if (errno == EPIPE || ret == -1 || (ret == 0 && errno != 0)) {
         conn_disconnect(conn);
     }
 

src/connection/mqtt_client.c

@@ -33,6 +33,7 @@
 #include "connection/mqtt_client.h"
 #include "event/event.h"
 #include "utils/asprintf.h"
+#include "utils/time.h"
 #include "utils/utextend.h"
 #include "utils/uthash.h"
 #include "utils/utlist.h"
@@ -68,6 +69,7 @@ typedef struct {
 typedef enum {
     TASK_PUB,
     TASK_SUB,
+    TASK_UNSUB,
     TASK_RECV,
 } task_kind_e;
 
@@ -133,7 +135,8 @@ static inline void    tasks_free(task_t *tasks);
 static void           task_cb(void *arg);
 static void           task_handle_pub(task_t *task, neu_mqtt_client_t *client);
 static void           task_handle_sub(task_t *task, neu_mqtt_client_t *client);
-static void           task_handle_recv(task_t *task, neu_mqtt_client_t *client);
+static void task_handle_unsub(task_t *task, neu_mqtt_client_t *client);
+static void task_handle_recv(task_t *task, neu_mqtt_client_t *client);
 
 static subscription_t *       subscription_new(neu_mqtt_client_t *client,
                                                neu_mqtt_qos_e qos, const char *topic,
@@ -221,6 +224,8 @@ static void task_cb(void *arg)
         task_handle_pub(task, client);
     } else if (TASK_SUB == task->kind) {
         task_handle_sub(task, client);
+    } else if (TASK_UNSUB == task->kind) {
+        task_handle_unsub(task, client);
     } else if (TASK_RECV == task->kind) {
         task_handle_recv(task, client);
     } else {
@@ -295,6 +300,41 @@ end:
     return;
 }
 
+static void task_handle_unsub(task_t *task, neu_mqtt_client_t *client)
+{
+    int      rv  = 0;
+    nng_msg *msg = NULL;
+    nng_aio *aio = task->aio;
+
+    if (0 != (rv = nng_aio_result(aio))) {
+        log(error, "send UNSUBSCRIBE error: %s", nng_strerror(rv));
+        goto end;
+    }
+
+    msg = nng_aio_get_msg(aio);
+    if (NULL == msg) {
+        rv = -1;
+        goto end;
+    }
+
+    nng_mqtt_packet_type pt = nng_mqtt_msg_get_packet_type(msg);
+    if (NNG_MQTT_UNSUBACK != pt) {
+        log(error, "recv packet type:%d, expect UNSUBACK packet", pt);
+        rv = -1;
+        goto end;
+    }
+
+end:
+    if (task->sub) {
+        log(info, "unsub [%s, QoS%d], %s", task->sub->topic, task->sub->qos,
+            (0 == rv) ? "OK" : "FAIL");
+    } else {
+        log(info, "unsub %s", (0 == rv) ? "OK" : "FAIL");
+    }
+    nng_msg_free(msg);
+    return;
+}
+
 static void task_handle_recv(task_t *task, neu_mqtt_client_t *client)
 {
     (void) client;
@@ -535,7 +575,7 @@ static inline void client_free_task(neu_mqtt_client_t *client, task_t *task)
 {
     nng_aio_set_msg(task->aio, NULL);
 
-    if (TASK_SUB == task->kind) {
+    if (TASK_SUB == task->kind || TASK_UNSUB == task->kind) {
         subscription_free(task->sub);
     } else if (TASK_RECV == task->kind) {
         subscription_free(task->recv.sub);
@@ -569,6 +609,16 @@ static inline void client_add_subscription(neu_mqtt_client_t *client,
     HASH_ADD_STR(client->subscriptions, topic, sub);
 }
 
+static inline void client_del_subscription(neu_mqtt_client_t *client,
+                                           subscription_t *   sub)
+{
+    HASH_DEL(client->subscriptions, sub);
+    if (sub->ack) {
+        client->suback_count -= 1;
+    }
+    subscription_free(sub);
+}
+
 static int client_send_sub_msg(neu_mqtt_client_t *client,
                                subscription_t *   subscription)
 {
@@ -606,6 +656,38 @@ static int client_send_sub_msg(neu_mqtt_client_t *client,
     return 0;
 }
 
+static int client_send_unsub_msg(neu_mqtt_client_t *client,
+                                 subscription_t *   subscription)
+{
+    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 mqtt_topic = { .buf    = (uint8_t *) subscription->topic,
+                                  .length = strlen(subscription->topic) };
+
+    nng_mqtt_msg_set_packet_type(sub_msg, NNG_MQTT_UNSUBSCRIBE);
+    nng_mqtt_msg_set_unsubscribe_topics(sub_msg, &mqtt_topic, 1);
+
+    task_t *task = client_alloc_task(client);
+    if (NULL == task) {
+        log(error, "client_alloc_task fail");
+        nng_msg_free(sub_msg);
+        return -1;
+    }
+
+    task->kind = TASK_UNSUB;
+    task->sub  = subscription ? subscription_ref(subscription) : NULL;
+    nng_aio_set_msg(task->aio, sub_msg);
+    nng_send_aio(client->sock, task->aio);
+
+    return 0;
+}
+
 static inline void client_start_recv(neu_mqtt_client_t *client)
 {
     if (client->recv_aio && !client->receiving) {
@@ -1163,7 +1245,7 @@ int neu_mqtt_client_close(neu_mqtt_client_t *client)
     // wait for all tasks
     while (client_task_free_list_len(client) != client->task_count) {
         nng_mtx_unlock(client->mtx);
-        nng_msleep(100);
+        neu_msleep(100);
         nng_mtx_lock(client->mtx);
     }
     client->open = false;
@@ -1261,3 +1343,43 @@ error:
     subscription_free(subscription);
     return -1;
 }
+
+int neu_mqtt_client_unsubscribe(neu_mqtt_client_t *client, const char *topic)
+{
+    int             rv           = 0;
+    subscription_t *subscription = NULL;
+
+    nng_mtx_lock(client->mtx);
+    if (NULL == client->recv_aio) {
+        if ((rv = nng_aio_alloc(&client->recv_aio, recv_cb, client)) != 0) {
+            log(error, "nng_aio_alloc fail: %s", nng_strerror(rv));
+            goto error;
+        }
+        if (client->connected) {
+            // connect_cb already fired, then start receiving
+            client_start_recv(client);
+        }
+    }
+
+    HASH_FIND_STR(client->subscriptions, topic, subscription);
+
+    if (client->connected) {
+        // send a unsubscribe message even subscription is NULL
+        if (client_send_unsub_msg(client, subscription) < 0) {
+            log(error, "client_send_unsub_msg fail");
+            goto error;
+        }
+    }
+
+    if (subscription) {
+        client_del_subscription(client, subscription);
+    }
+
+    nng_mtx_unlock(client->mtx);
+
+    return 0;
+
+error:
+    nng_mtx_unlock(client->mtx);
+    return -1;
+}

src/core/manager.c

@@ -97,8 +97,9 @@ neu_manager_t *neu_manager_create()
     rv = nng_listen(manager->socket, url, NULL, 0);
     assert(rv == 0);
 
-    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 2048);
-    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 2048);
+    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 8292);
+    nng_socket_set_ms(manager->socket, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(manager->socket, NNG_OPT_RECVFD, &param.fd);
     manager->loop = neu_event_add_io(manager->events, param);
 
@@ -482,6 +483,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             forward_msg(manager, msg, cmd->app);
             manager_storage_subscribe(manager, cmd->app, cmd->driver,
                                       cmd->group, cmd->params);
+        } else {
+            free(cmd->params);
         }
 
         header->type = NEU_RESP_ERROR;
@@ -729,6 +732,7 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg,
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
         nlog_debug("forward msg to pipe %d", pipe.id);
     } else {
+        nlog_warn("forward msg to pipe %d fail", pipe.id);
         nng_msg_free(out_msg);
     }
 }
@@ -744,6 +748,7 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
         nlog_debug("forward msg to %s", node);
     } else {
+        nlog_warn("forward msg to pipe (%s)%d fail", node, pipe.id);
         nng_msg_free(out_msg);
     }
 }

version

@@ -1 +1 @@
-2.4.4
+2.4.6