Merge pull request #1588 from fengzeroz/v2.5

version: 2.5.3

.github/workflows/check.yml

@@ -103,4 +103,38 @@ jobs:
             ft/reports/log.html
             ft/reports/output.xml
             ft/reports/report.html
-            build/neuron.log
+            build/neuron.log
+
+  pyft:
+    runs-on: ubuntu-20.04
+    container: neugates/build:x86_64-main
+
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: recursive
+      
+      - name: compile
+        run: |
+          git config --global --add safe.directory $(pwd)
+          mkdir -p build && cd build
+          cmake .. -DCMAKE_TOOLCHAIN_FILE=../cmake/x86_64-linux-gnu.cmake
+          make -j4
+
+      - name: fetch neu-ft
+        uses: actions/checkout@v2
+        with:
+            repository: neugates/neu-ft
+            path: neu-ft
+
+      - name: function test
+        run: |
+          pip install -U pytest
+          pip install requests
+          pytest -s neu-ft/neuron
+
+      - uses: actions/upload-artifact@v3
+        with:
+          name: pyft_result
+          path: |
+            neu-ft/neuron/report

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.5.3
+  VERSION: 2.5.4
   UI_VERSION: 2.5.3
   BRANCH: v2.5
 

CMakeLists.txt

@@ -36,6 +36,7 @@ if(USE_GCOV)
   SET(GCC_COVERAGE_COMPILE_FLAGS "-fprofile-arcs -ftest-coverage")
   SET(GCC_COVERAGE_LINK_FLAGS "-lgcov")
   SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${GCC_COVERAGE_COMPILE_FLAGS}")
+  SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${GCC_COVERAGE_COMPILE_FLAGS}")
   SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_LINKER_FLAGS} ${GCC_COVERAGE_LINK_FLAGS}")
 endif(USE_GCOV)
 
@@ -150,6 +151,7 @@ target_link_options(neuron PRIVATE "LINKER:--dynamic-list-data")
 #copy file for run
 file(COPY ${CMAKE_SOURCE_DIR}/zlog.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(COPY ${CMAKE_SOURCE_DIR}/dev.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
+file(COPY ${CMAKE_SOURCE_DIR}/neuron.json DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(COPY ${CMAKE_SOURCE_DIR}/.gitkeep DESTINATION ${CMAKE_BINARY_DIR}/logs)
 file(COPY ${CMAKE_SOURCE_DIR}/.gitkeep DESTINATION ${CMAKE_BINARY_DIR}/persistence)
 file(COPY ${CMAKE_SOURCE_DIR}/neuron.key DESTINATION ${CMAKE_BINARY_DIR}/config)

ft/modbus_rtu_test.robot

@@ -48,6 +48,13 @@ ${input_bit_1_name_wrong}           {"name": "input_bit", "address": "1!100001",
 ${input_bit_1_attribute_unmatch}    {"name": "input_bit_1", "address": "1!100001", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_BIT}}
 ${input_bit_1_type_unmatch}         {"name": "input_bit_1", "address": "1!100001", "attribute": ${TAG_ATTRIBUTE_READ}, "type": ${TAG_TYPE_INT16}}
 
+${hold_int16_1}     {"name": "hold_int16_1", "address": "1!400001", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_INT16}}
+${hold_int16_2}     {"name": "hold_int16_2", "address": "1!400002", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_INT16}}
+${hold_int16_3}     {"name": "hold_int16_3", "address": "1!400003", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_INT16}}
+${hold_string_tags_1}    {"name": "hold_string_tags_1", "address": "1!40001.4", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_STRING}}
+${hold_string_tags_2}    {"name": "hold_string_tags_2", "address": "1!40003.4", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_STRING}}
+${hold_string_tags_3}    {"name": "hold_string_tags_3", "address": "1!40005.4", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_STRING}}
+
 *** Test Cases ***
 Set a node with right settings, it should be success.
     [Template]	Set a ${node} with right ${node_settings}, it will be success.
@@ -146,6 +153,12 @@ Update a tag with unmatch address and types and attribute, it should be failure.
     ${input_bit_1}	${input_bit_1_type_unmatch}	${modbus_rtu_node}	200	${NEU_ERR_TAG_TYPE_NOT_SUPPORT}
     ${input_bit_1}	${input_bit_1_name_wrong}	${modbus_rtu_node}	200	${NEU_ERR_TAG_NOT_EXIST}
 
+Write 3 tags 1 time, it should return success
+    [Template]  Write and Read ${tag_data1}, ${tag_data2} and ${tag_data3} named ${tag1}, ${tag2} and ${tag3} from ${node}, using ${check} to check the ${value1}, ${value2} and ${value3}, it will be success.
+    ${hold_int16_1}     ${hold_int16_2}    ${hold_int16_3}    hold_int16_1    hold_int16_2    hold_int16_3     ${modbus_rtu_node}	  Compare Tag Value As Int        1    2    3
+	${coil_bit_1}       ${coil_bit_2}      ${coil_bit_3}      coil_bit_1      coil_bit_2      coil_bit_3       ${modbus_rtu_node}    Compare Tag Value As Int        1    0    1
+	${hold_string_tags_1}     ${hold_string_tags_2}    ${hold_string_tags_3}    hold_string_tags_1    hold_string_tags_2    hold_string_tags_3    ${modbus_rtu_node}	  Compare Tag Value As Strings        "a"    "b"    "c"
+
 Write a value to a tag that don't exit in plc or neuron, it will be failure.
     [Template]	Write a ${value} to a ${tag} with named ${tag_name} that don't exist in neuron or plc from ${node}, it will be failure.
     12	${input_bit_1}	input	${modbus_rtu_node}

ft/modbus_tcp_test.robot

@@ -69,6 +69,13 @@ ${hold_double_decimal}    {"name": "hold_double_decimal", "address": "1!400027",
 ${hold_float_precision}      {"name": "hold_float_precision", "address": "1!400030", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_FLOAT}, "precision": 2}
 ${hold_double_precision}     {"name": "hold_double_precision", "address": "1!400040", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_DOUBLE}, "precision": 2}
 
+${hold_int16_1}     {"name": "hold_int16_1", "address": "1!400001", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_INT16}}
+${hold_int16_2}     {"name": "hold_int16_2", "address": "1!400002", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_INT16}}
+${hold_int16_3}     {"name": "hold_int16_3", "address": "1!400003", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_INT16}}
+${hold_string_tags_1}    {"name": "hold_string_tags_1", "address": "1!40001.4", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_STRING}}
+${hold_string_tags_2}    {"name": "hold_string_tags_2", "address": "1!40003.4", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_STRING}}
+${hold_string_tags_3}    {"name": "hold_string_tags_3", "address": "1!40005.4", "attribute": ${TAG_ATTRIBUTE_RW}, "type": ${TAG_TYPE_STRING}}
+
 *** Test Cases ***
 Set a node with right settings, it should be success.
 	[Template]	Set a ${node} with right ${node_settings}, it will be success.
@@ -167,6 +174,12 @@ Update a tag with right address and types and attribute, it will be success
 	${hold_int16}	${hold_int16_address_update}	hold_int16	${modbus_tcp_node}	Compare Tag Value As Int  120
 	${hold_int16}	${hold_int16_type_update}	    hold_int16	${modbus_tcp_node}	Compare Tag Value As Int  120
 
+Write 3 tags 1 time, it should return success
+    [Template]  Write and Read ${tag_data1}, ${tag_data2} and ${tag_data3} named ${tag1}, ${tag2} and ${tag3} from ${node}, using ${check} to check the ${value1}, ${value2} and ${value3}, it will be success.
+    ${hold_int16_1}     ${hold_int16_2}    ${hold_int16_3}    hold_int16_1    hold_int16_2    hold_int16_3     ${modbus_tcp_node}	  Compare Tag Value As Int        1    2    3
+	${coil_bit_1}       ${coil_bit_2}      ${coil_bit_3}      coil_bit_1      coil_bit_2      coil_bit_3       ${modbus_tcp_node}    Compare Tag Value As Int        1    0    1
+	${hold_string_tags_1}     ${hold_string_tags_2}    ${hold_string_tags_3}    hold_string_tags_1    hold_string_tags_2    hold_string_tags_3    ${modbus_tcp_node}	  Compare Tag Value As Strings        "a"    "b"    "c"
+
 Add a tag with decimal, it should return success
 	[Template]	Add a ${tag} with right address and type and attribute from ${node}, it will be success.
 	${hold_int16_decimal}	${modbus_tcp_node}

ft/resource/api.resource

@@ -186,6 +186,12 @@ Write Tags
   ${res} =            Object  response body
   [RETURN]            ${res}[0]
 
+Write 3 Tags 1 Time
+  [Arguments]         ${node}    ${group}    ${tag1}     ${value1}    ${tag2}     ${value2}    ${tag3}     ${value3}
+  POST                /api/v2/write/tags           {"node": "${node}", "group": "${group}", "tags": [{"tag": "${tag1}", "value": ${value1}}, {"tag": "${tag2}", "value": ${value2}}, {"tag": "${tag3}", "value": ${value3}}]}
+  ${res} =            Object  response body
+  [RETURN]            ${res}[0]
+
 Add Template
   [Arguments]   ${name}   ${plugin}   ${groups}
   ${data} =     Evaluate              json.dumps(${groups})   json

ft/resource/driver.resource

@@ -293,3 +293,37 @@ Write and read a ${value} to a ${tag} named ${tag_name} when the ${node} is stop
 	Check Tag Error As Int  ${res}[tags]  ${tag_name}  ${NEU_ERR_PLUGIN_NOT_RUNNING}
 
 	[Teardown]	Del Group  ${node}  group
+
+Write and Read ${tag_data1}, ${tag_data2} and ${tag_data3} named ${tag1}, ${tag2} and ${tag3} from ${node}, using ${check} to check the ${value1}, ${value2} and ${value3}, it will be success.
+	[Tags]	Driver 	18
+
+	${res}=	Add Group  ${node}  group  100
+	Check Response Status  ${res}  200
+	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
+
+	${res}=	Add Tags  ${node}  group  ${tag_data1}
+	Check Response Status  ${res}  200
+	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
+
+	${res}=	Add Tags  ${node}  group  ${tag_data2}
+	Check Response Status  ${res}  200
+	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
+
+	${res}=	Add Tags  ${node}  group  ${tag_data3}
+	Check Response Status  ${res}  200
+	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
+
+	Sleep	290ms
+	${res}=	Write 3 Tags 1 Time  ${node}  group  ${tag1}  ${value1}  ${tag2}  ${value2}  ${tag3}  ${value3}
+	Check Response Status  ${res}  200
+	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
+
+	Sleep	100ms
+	${res}=	Read Tags  ${node}  group
+	Check Response Status  ${res}  200
+
+	Run Keyword	${check}	${res}[tags]	${tag1}	${value1}
+	Run Keyword	${check}	${res}[tags]	${tag2}	${value2}
+	Run Keyword	${check}	${res}[tags]	${tag3}	${value3}
+
+	[Teardown]	Del Group    ${node} 	group

ft/resource/error.resource

@@ -94,3 +94,5 @@ ${NEU_ERR_PLUGIN_NOT_FOUND}                     3014
 ${NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE}           3015
 ${NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE}          3016
 ${NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS}        3017
+
+${NEU_ERR_MQTT_SUBSCRIBE_FAILURE}               4010

ft/subscribe_test.robot

@@ -24,6 +24,13 @@ North APP subscribe existent group, it should return success
 	Check Response Status  ${res}  200
 	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
 
+North MQTT APP subscribe with empty topic, it should fail.
+	POST                /api/v2/subscribe  {"app": "mqtt", "driver": "modbus-node", "group": "group", "params":{"topic":""}}
+	${res} =            Object  response body
+
+	Check Response Status           ${res}      200
+	Check Error Code                ${res}      ${NEU_ERR_MQTT_SUBSCRIBE_FAILURE}
+
 Query the subscribers of the group, it should return all nodes subscribed to this group
 	${res} =                  Get Subscribe Group		mqtt
 	Check Response Status     ${res}                          200

include/neuron/define.h

@@ -42,6 +42,7 @@
 
 extern int  default_log_level;
 extern bool disable_jwt;
+extern char host_port[24];
 
 typedef enum neu_plugin_kind {
     NEU_PLUGIN_KIND_STATIC = 0,

include/neuron/plugin.h

@@ -72,7 +72,7 @@ typedef struct {
 typedef struct neu_plugin_intf_funs {
     neu_plugin_t *(*open)(void);
     int (*close)(neu_plugin_t *plugin);
-    int (*init)(neu_plugin_t *plugin);
+    int (*init)(neu_plugin_t *plugin, bool load);
     int (*uninit)(neu_plugin_t *plugin);
     int (*start)(neu_plugin_t *plugin);
     int (*stop)(neu_plugin_t *plugin);
@@ -90,6 +90,14 @@ typedef struct neu_plugin_intf_funs {
                 neu_plugin_t *plugin, void *req,
                 UT_array *tag_values); // UT_array {neu_datatag_t, neu_value_u}
             neu_plugin_tag_validator_t tag_validator;
+
+            int (*load_tags)(
+                neu_plugin_t *plugin, const char *group, neu_datatag_t *tags,
+                int n_tag); // create tags using data from the database
+            int (*add_tags)(neu_plugin_t *plugin, const char *group,
+                            neu_datatag_t *tags,
+                            int            n_tag); // create tags by API
+            int (*del_tags)(neu_plugin_t *plugin, int n_tag);
         } driver;
     };
 

neuron.json

@@ -0,0 +1,5 @@
+{
+	"ip": "0.0.0.0",
+	"port": 7000,
+	"disable_auth": 0
+}

package-sdk.sh

@@ -53,6 +53,7 @@ cp zlog.conf ${package_name}/config/
 cp dev.conf ${package_name}/config/
 cp neuron.key ${package_name}/config/
 cp neuron.pem ${package_name}/config/
+cp neuron.json ${package_name}/config/
 cp default_plugins.json ${package_name}/config/
 
 cp build/plugins/schema/ekuiper.json \

plugins/ekuiper/plugin_ekuiper.c

@@ -73,8 +73,9 @@ static void pipe_rm_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     nng_mtx_unlock(plugin->mtx);
 }
 
-static int ekuiper_plugin_init(neu_plugin_t *plugin)
+static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     int      rv       = 0;
     nng_aio *recv_aio = NULL;
 

plugins/modbus/modbus-rtu.json

@@ -80,7 +80,7 @@
 		"name": "Maximum Retry Times",
 		"name_zh": "最大重试次数",
 		"description": "The maximum number of retries after a failed attempt to send a read command",
-		"description_zh": "发送读取指令失败后最大重试次数",
+		"description_zh": "发送读指令失败后最大重试次数",
 		"attribute": "required",
 		"type": "int",
 		"default": 0,
@@ -93,7 +93,7 @@
 		"name": "Retry Interval",
 		"name_zh": "指令重新发送间隔",
 		"description": "Resend reading instruction interval(ms) after a failed attempt to send a read command",
-		"description_zh": "发送读取指令失败后重新发送读指令时间间隔，单位为毫秒",
+		"description_zh": "发送读指令失败后重新发送读指令时间间隔，单位为毫秒",
 		"attribute": "required",
 		"type": "int",
 		"default": 0,
@@ -106,7 +106,7 @@
 		"name": "Send Interval",
 		"name_zh": "指令发送间隔",
 		"description": "Send reading instruction interval(ms)",
-		"description_zh": "发送读指令时间间隔，单位为毫秒",
+		"description_zh": "发送指令时间间隔，单位为毫秒",
 		"attribute": "required",
 		"type": "int",
 		"default": 20,

plugins/modbus/modbus-tcp.json

@@ -92,7 +92,7 @@
 		"name": "Maximum Retry Times",
 		"name_zh": "最大重试次数",
 		"description": "The maximum number of retries after a failed attempt to send a read command",
-		"description_zh": "发送读取指令失败后最大重试次数",
+		"description_zh": "发送读指令失败后最大重试次数",
 		"attribute": "required",
 		"type": "int",
 		"default": 0,
@@ -105,7 +105,7 @@
 		"name": "Retry Interval",
 		"name_zh": "指令重新发送间隔",
 		"description": "Resend reading instruction interval(ms) after a failed attempt to send a read command",
-		"description_zh": "发送读取指令失败后重新发送读指令时间间隔，单位为毫秒",
+		"description_zh": "发送读指令失败后重新发送读指令时间间隔，单位为毫秒",
 		"attribute": "required",
 		"type": "int",
 		"default": 0,
@@ -118,7 +118,7 @@
 		"name": "Send Interval",
 		"name_zh": "指令发送间隔",
 		"description": "Send reading instruction interval(ms)",
-		"description_zh": "发送读指令时间间隔，单位为毫秒",
+		"description_zh": "发送指令时间间隔，单位为毫秒",
 		"attribute": "required",
 		"type": "int",
 		"default": 20,

plugins/modbus/modbus_point.c

@@ -27,10 +27,13 @@ struct modbus_sort_ctx {
     uint16_t end;
 };
 
-static __thread uint16_t modbus_read_max_byte = 255;
+static __thread uint16_t modbus_read_max_byte = 250;
 
 static int  tag_cmp(neu_tag_sort_elem_t *tag1, neu_tag_sort_elem_t *tag2);
 static bool tag_sort(neu_tag_sort_t *sort, void *tag, void *tag_to_be_sorted);
+static int  tag_cmp_write(neu_tag_sort_elem_t *tag1, neu_tag_sort_elem_t *tag2);
+static bool tag_sort_write(neu_tag_sort_t *sort, void *tag,
+                           void *tag_to_be_sorted);
 
 int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point)
 {
@@ -165,6 +168,15 @@ int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point)
     return ret;
 }
 
+int modbus_write_tag_to_point(const neu_plugin_tag_value_t *tag,
+                              modbus_point_write_t *        point)
+{
+    int ret      = NEU_ERR_SUCCESS;
+    ret          = modbus_tag_to_point(tag->tag, &point->point);
+    point->value = tag->value;
+    return ret;
+}
+
 modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte)
 {
     modbus_read_max_byte          = max_byte;
@@ -193,6 +205,112 @@ modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte)
     return sort_result;
 }
 
+int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option)
+{
+    int n = 0;
+    switch (type) {
+    case NEU_TYPE_UINT16:
+    case NEU_TYPE_INT16:
+        n          = sizeof(uint16_t);
+        value->u16 = htons(value->u16);
+        break;
+    case NEU_TYPE_FLOAT:
+    case NEU_TYPE_UINT32:
+    case NEU_TYPE_INT32:
+        n          = sizeof(uint32_t);
+        value->u32 = htonl(value->u32);
+        break;
+
+    case NEU_TYPE_DOUBLE:
+    case NEU_TYPE_INT64:
+    case NEU_TYPE_UINT64:
+        n          = sizeof(uint64_t);
+        value->u64 = neu_htonll(value->u64);
+        break;
+    case NEU_TYPE_BIT: {
+        n = sizeof(uint8_t);
+        break;
+    }
+    case NEU_TYPE_STRING: {
+        n = option.string.length;
+        switch (option.string.type) {
+        case NEU_DATATAG_STRING_TYPE_H:
+            break;
+        case NEU_DATATAG_STRING_TYPE_L:
+            neu_datatag_string_ltoh(value->str, option.string.length);
+            break;
+        case NEU_DATATAG_STRING_TYPE_D:
+            break;
+        case NEU_DATATAG_STRING_TYPE_E:
+            break;
+        }
+        break;
+    }
+    default:
+        assert(1 == 0);
+        break;
+    }
+    return n;
+}
+
+modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags)
+{
+    neu_tag_sort_result_t *result =
+        neu_tag_sort(tags, tag_sort_write, tag_cmp_write);
+
+    modbus_write_cmd_sort_t *sort_result =
+        calloc(1, sizeof(modbus_write_cmd_sort_t));
+    sort_result->n_cmd = result->n_sort;
+    sort_result->cmd   = calloc(result->n_sort, sizeof(modbus_write_cmd_t));
+    for (uint16_t i = 0; i < result->n_sort; i++) {
+        modbus_point_write_t *tag =
+            *(modbus_point_write_t **) utarray_front(result->sorts[i].tags);
+        struct modbus_sort_ctx *ctx = result->sorts[i].info.context;
+
+        int num_tags              = utarray_len(result->sorts[i].tags);
+        sort_result->cmd[i].bytes = calloc(num_tags, sizeof(neu_value_u));
+        int      n_byte = 0, n_byte_tag = 0;
+        uint8_t *data_bit = calloc((num_tags + 7) / 8, sizeof(uint8_t));
+        int      k        = 0;
+        utarray_foreach(result->sorts[i].tags, modbus_point_write_t **, tag_s)
+        {
+            if ((*tag_s)->point.area == MODBUS_AREA_COIL) {
+                n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
+                                        (*tag_s)->point.option);
+                data_bit[k / 8] += ((*tag_s)->value.i8) << k % 8;
+                n_byte += n_byte_tag;
+                k++;
+            } else {
+                n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
+                                        (*tag_s)->point.option);
+                memcpy(sort_result->cmd[i].bytes +
+                           2 *
+                               ((*tag_s)->point.start_address -
+                                tag->point.start_address),
+                       &((*tag_s)->value), n_byte_tag);
+                n_byte += n_byte_tag;
+            }
+        }
+        if ((*(modbus_point_write_t **) utarray_front(result->sorts[i].tags))
+                ->point.area == MODBUS_AREA_COIL) {
+            memcpy(sort_result->cmd[i].bytes, data_bit, (k + 7) / 8);
+        }
+
+        sort_result->cmd[i].tags     = utarray_clone(result->sorts[i].tags);
+        sort_result->cmd[i].slave_id = tag->point.slave_id;
+        sort_result->cmd[i].area     = tag->point.area;
+        sort_result->cmd[i].start_address = tag->point.start_address;
+        sort_result->cmd[i].n_register    = ctx->end - ctx->start;
+        sort_result->cmd[i].n_byte        = n_byte;
+
+        free(data_bit);
+        free(result->sorts[i].info.context);
+    }
+
+    neu_tag_sort_free(result);
+    return sort_result;
+}
+
 void modbus_tag_sort_free(modbus_read_cmd_sort_t *cs)
 {
     for (uint16_t i = 0; i < cs->n_cmd; i++) {
@@ -266,7 +384,7 @@ static bool tag_sort(neu_tag_sort_t *sort, void *tag, void *tag_to_be_sorted)
     switch (t1->area) {
     case MODBUS_AREA_COIL:
     case MODBUS_AREA_INPUT:
-        if ((ctx->end - ctx->start) / 8 >= modbus_read_max_byte - 1) {
+        if ((ctx->end - ctx->start + 7) / 8 >= modbus_read_max_byte) {
             return false;
         }
         break;
@@ -288,3 +406,90 @@ static bool tag_sort(neu_tag_sort_t *sort, void *tag, void *tag_to_be_sorted)
 
     return true;
 }
+
+static int tag_cmp_write(neu_tag_sort_elem_t *tag1, neu_tag_sort_elem_t *tag2)
+{
+    modbus_point_write_t *p_t1 = (modbus_point_write_t *) tag1->tag;
+    modbus_point_write_t *p_t2 = (modbus_point_write_t *) tag2->tag;
+
+    if (p_t1->point.slave_id > p_t2->point.slave_id) {
+        return 1;
+    } else if (p_t1->point.slave_id < p_t2->point.slave_id) {
+        return -1;
+    }
+
+    if (p_t1->point.area > p_t2->point.area) {
+        return 1;
+    } else if (p_t1->point.area < p_t2->point.area) {
+        return -1;
+    }
+
+    if (p_t1->point.start_address > p_t2->point.start_address) {
+        return 1;
+    } else if (p_t1->point.start_address < p_t2->point.start_address) {
+        return -1;
+    }
+
+    if (p_t1->point.n_register > p_t2->point.n_register) {
+        return 1;
+    } else if (p_t1->point.n_register < p_t2->point.n_register) {
+        return -1;
+    }
+
+    return 0;
+}
+
+static bool tag_sort_write(neu_tag_sort_t *sort, void *tag,
+                           void *tag_to_be_sorted)
+{
+    modbus_point_write_t *  t1  = (modbus_point_write_t *) tag;
+    modbus_point_write_t *  t2  = (modbus_point_write_t *) tag_to_be_sorted;
+    struct modbus_sort_ctx *ctx = NULL;
+
+    if (sort->info.context == NULL) {
+        sort->info.context = calloc(1, sizeof(struct modbus_sort_ctx));
+        ctx                = (struct modbus_sort_ctx *) sort->info.context;
+        ctx->start         = t1->point.start_address;
+        ctx->end           = t1->point.start_address + t1->point.n_register;
+        return true;
+    }
+
+    ctx = (struct modbus_sort_ctx *) sort->info.context;
+
+    if (t1->point.slave_id != t2->point.slave_id) {
+        return false;
+    }
+
+    if (t1->point.area != t2->point.area) {
+        return false;
+    }
+
+    if (t2->point.start_address > ctx->end) {
+        return false;
+    }
+
+    switch (t1->point.area) {
+    case MODBUS_AREA_COIL:
+    case MODBUS_AREA_INPUT:
+        if ((ctx->end - ctx->start) / 8 >= modbus_read_max_byte - 1) {
+            return false;
+        }
+        break;
+    case MODBUS_AREA_INPUT_REGISTER:
+    case MODBUS_AREA_HOLD_REGISTER: {
+        uint16_t now_bytes = (ctx->end - ctx->start) * 2;
+        uint16_t add_now   = now_bytes + t2->point.n_register * 2;
+        if (add_now >= modbus_read_max_byte) {
+            return false;
+        }
+
+        break;
+    }
+    }
+
+    if (t2->point.start_address + t2->point.n_register > ctx->end) {
+        ctx->end = t2->point.start_address + t2->point.n_register;
+    }
+
+    return true;
+}

plugins/modbus/modbus_point.h

@@ -40,7 +40,14 @@ typedef struct modbus_point {
     char                      name[NEU_TAG_NAME_LEN];
 } modbus_point_t;
 
+typedef struct modbus_point_write {
+    modbus_point_t point;
+    neu_value_u    value;
+} modbus_point_write_t;
+
 int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point);
+int modbus_write_tag_to_point(const neu_plugin_tag_value_t *tag,
+                              modbus_point_write_t *        point);
 
 typedef struct modbus_read_cmd {
     uint8_t       slave_id;
@@ -56,8 +63,25 @@ typedef struct modbus_read_cmd_sort {
     modbus_read_cmd_t *cmd;
 } modbus_read_cmd_sort_t;
 
-modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte);
-void                    modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
+typedef struct modbus_write_cmd {
+    uint8_t       slave_id;
+    modbus_area_e area;
+    uint16_t      start_address;
+    uint16_t      n_register;
+    uint8_t       n_byte;
+    uint8_t *     bytes;
+
+    UT_array *tags;
+} modbus_write_cmd_t;
+
+typedef struct modbus_write_cmd_sort {
+    uint16_t            n_cmd;
+    modbus_write_cmd_t *cmd;
+} modbus_write_cmd_sort_t;
+
+modbus_read_cmd_sort_t * modbus_tag_sort(UT_array *tags, uint16_t max_byte);
+modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags);
+void                     modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
 
 #ifdef __cplusplus
 }

plugins/modbus/modbus_req.c

@@ -29,6 +29,11 @@ struct modbus_group_data {
     modbus_read_cmd_sort_t *cmd_sort;
 };
 
+struct modbus_write_tags_data {
+    UT_array *               tags;
+    modbus_write_cmd_sort_t *cmd_sort;
+};
+
 static void plugin_group_free(neu_plugin_group_t *pgp);
 static int  process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
                                  uint16_t response_size);
@@ -101,7 +106,7 @@ int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes)
 
     plog_send_protocol(plugin, bytes, n_byte);
 
-    if (plugin->is_server && plugin->protocol == MODBUS_PROTOCOL_TCP) {
+    if (plugin->is_server) {
         ret = neu_conn_tcp_server_send(plugin->conn, plugin->client_fd, bytes,
                                        n_byte);
     } else {
@@ -445,7 +450,7 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
 }
 
 int modbus_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
-                 neu_value_u value)
+                 neu_value_u value, bool response)
 {
     modbus_point_t point = { 0 };
     int            ret   = modbus_tag_to_point(tag, &point);
@@ -498,12 +503,83 @@ int modbus_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
     uint16_t response_size = 0;
     ret = modbus_stack_write(plugin->stack, req, point.slave_id, point.area,
                              point.start_address, point.n_register, value.bytes,
-                             n_byte, &response_size);
+                             n_byte, &response_size, response);
     if (ret > 0) {
         process_protocol_buf(plugin, point.slave_id, response_size);
     }
 
-    return 0;
+    return ret;
+}
+
+int modbus_write_tag(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
+                     neu_value_u value)
+{
+    int ret = 0;
+    ret     = modbus_write(plugin, req, tag, value, true);
+    return ret;
+}
+
+int modbus_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
+{
+    struct modbus_write_tags_data *gtags = NULL;
+    int                            ret   = 0;
+    int                            rv    = 0;
+
+    gtags = calloc(1, sizeof(struct modbus_write_tags_data));
+
+    utarray_new(gtags->tags, &ut_ptr_icd);
+    utarray_foreach(tags, neu_plugin_tag_value_t *, tag)
+    {
+        modbus_point_write_t *p = calloc(1, sizeof(modbus_point_write_t));
+        ret                     = modbus_write_tag_to_point(tag, p);
+        assert(ret == 0);
+
+        utarray_push_back(gtags->tags, &p);
+    }
+    gtags->cmd_sort = modbus_write_tags_sort(gtags->tags);
+    for (uint16_t i = 0; i < gtags->cmd_sort->n_cmd; i++) {
+        uint16_t response_size = 0;
+
+        ret = modbus_stack_write(
+            plugin->stack, req, gtags->cmd_sort->cmd[i].slave_id,
+            gtags->cmd_sort->cmd[i].area, gtags->cmd_sort->cmd[i].start_address,
+            gtags->cmd_sort->cmd[i].n_register, gtags->cmd_sort->cmd[i].bytes,
+            gtags->cmd_sort->cmd[i].n_byte, &response_size, false);
+        if (ret > 0) {
+            process_protocol_buf(plugin, gtags->cmd_sort->cmd[i].slave_id,
+                                 response_size);
+        }
+
+        if (ret <= 0) {
+            rv = 1;
+        }
+        if (plugin->interval > 0) {
+            struct timespec t1 = { .tv_sec  = plugin->interval / 1000,
+                                   .tv_nsec = 1000 * 1000 *
+                                       (plugin->interval % 1000) };
+            struct timespec t2 = { 0 };
+            nanosleep(&t1, &t2);
+        }
+    }
+
+    if (rv == 0) {
+        plugin->common.adapter_callbacks->driver.write_response(
+            plugin->common.adapter, req, NEU_ERR_SUCCESS);
+    } else {
+        plugin->common.adapter_callbacks->driver.write_response(
+            plugin->common.adapter, req, NEU_ERR_PLUGIN_DISCONNECTED);
+    }
+
+    for (uint16_t i = 0; i < gtags->cmd_sort->n_cmd; i++) {
+        utarray_free(gtags->cmd_sort->cmd[i].tags);
+        free(gtags->cmd_sort->cmd[i].bytes);
+    }
+    free(gtags->cmd_sort->cmd);
+    free(gtags->cmd_sort);
+    utarray_foreach(gtags->tags, modbus_point_write_t **, tag) { free(*tag); }
+    utarray_free(gtags->tags);
+    free(gtags);
+    return ret;
 }
 
 int modbus_write_resp(void *ctx, void *req, int error)
@@ -598,7 +674,12 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
             }
         }
     } else if (plugin->protocol == MODBUS_PROTOCOL_RTU) {
-        ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
+        if (plugin->is_server) {
+            ret = neu_conn_tcp_server_recv(plugin->conn, plugin->client_fd,
+                                           recv_buf, response_size);
+        } else {
+            ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
+        }
 
         if (ret == 0 || ret == -1) {
             free(recv_buf);

plugins/modbus/modbus_req.h

@@ -58,7 +58,10 @@ int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes);
 int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
                         uint8_t *bytes, int error);
 int modbus_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
-                 neu_value_u value);
+                 neu_value_u value, bool response);
+int modbus_write_tag(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
+                     neu_value_u value);
+int modbus_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags);
 int modbus_write_resp(void *ctx, void *req, int error);
 
 #endif

plugins/modbus/modbus_rtu.c

@@ -29,7 +29,7 @@
 static neu_plugin_t *driver_open(void);
 
 static int driver_close(neu_plugin_t *plugin);
-static int driver_init(neu_plugin_t *plugin);
+static int driver_init(neu_plugin_t *plugin, bool load);
 static int driver_uninit(neu_plugin_t *plugin);
 static int driver_start(neu_plugin_t *plugin);
 static int driver_stop(neu_plugin_t *plugin);
@@ -42,6 +42,7 @@ static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
 static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
 static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
                         neu_value_u value);
+static int driver_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags);
 
 static const neu_plugin_intf_funs_t plugin_intf_funs = {
     .open    = driver_open,
@@ -57,7 +58,10 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
     .driver.group_timer   = driver_group_timer,
     .driver.write_tag     = driver_write,
     .driver.tag_validator = driver_tag_validator,
-    .driver.write_tags    = NULL,
+    .driver.write_tags    = driver_write_tags,
+    .driver.add_tags      = NULL,
+    .driver.load_tags     = NULL,
+    .driver.del_tags      = NULL,
 };
 
 const neu_plugin_module_t neu_plugin_module = {
@@ -94,8 +98,9 @@ static int driver_close(neu_plugin_t *plugin)
     return 0;
 }
 
-static int driver_init(neu_plugin_t *plugin)
+static int driver_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     plugin->protocol = MODBUS_PROTOCOL_RTU;
     plugin->events   = neu_event_new();
     plugin->stack    = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_RTU,
@@ -312,5 +317,10 @@ static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
 static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
                         neu_value_u value)
 {
-    return modbus_write(plugin, req, tag, value);
+    return modbus_write_tag(plugin, req, tag, value);
+}
+
+static int driver_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
+{
+    return modbus_write_tags(plugin, req, tags);
 }

plugins/modbus/modbus_stack.c

@@ -225,7 +225,7 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
 int modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,
                        enum modbus_area area, uint16_t start_address,
                        uint16_t n_reg, uint8_t *bytes, uint8_t n_byte,
-                       uint16_t *response_size)
+                       uint16_t *response_size, bool response)
 {
     static __thread neu_protocol_pack_buf_t pbuf     = { 0 };
     modbus_action_e                         m_action = MODBUS_ACTION_DEFAULT;
@@ -239,13 +239,21 @@ int modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,
 
     switch (area) {
     case MODBUS_AREA_COIL:
-        if (*bytes > 0) {
-            modbus_address_wrap(&pbuf, start_address, 0xff00, m_action);
+        if (n_byte > 1) {
+            n_reg = n_byte;
+            modbus_data_wrap(&pbuf, (n_byte + 7) / 8, bytes, m_action);
+            modbus_address_wrap(&pbuf, start_address, n_reg, m_action);
+            modbus_code_wrap(&pbuf, slave_id, MODBUS_WRITE_M_COIL);
+            break;
         } else {
-            modbus_address_wrap(&pbuf, start_address, 0, m_action);
+            if (*bytes > 0) {
+                modbus_address_wrap(&pbuf, start_address, 0xff00, m_action);
+            } else {
+                modbus_address_wrap(&pbuf, start_address, 0, m_action);
+            }
+            modbus_code_wrap(&pbuf, slave_id, MODBUS_WRITE_S_COIL);
+            break;
         }
-        modbus_code_wrap(&pbuf, slave_id, MODBUS_WRITE_S_COIL);
-        break;
     case MODBUS_AREA_HOLD_REGISTER:
         m_action = MODBUS_ACTION_HOLD_REG_WRITE;
         modbus_data_wrap(&pbuf, n_byte, bytes, m_action);
@@ -275,13 +283,17 @@ int modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,
     int ret = stack->send_fn(stack->ctx, neu_protocol_pack_buf_used_size(&pbuf),
                              neu_protocol_pack_buf_get(&pbuf));
     if (ret > 0) {
-        stack->write_resp(stack->ctx, req, NEU_ERR_SUCCESS);
-        plog_notice((neu_plugin_t *) stack->ctx, "send write req, %hhu!%hu",
-                    slave_id, start_address);
+        if (response) {
+            stack->write_resp(stack->ctx, req, NEU_ERR_SUCCESS);
+            plog_notice((neu_plugin_t *) stack->ctx, "send write req, %hhu!%hu",
+                        slave_id, start_address);
+        }
     } else {
-        stack->write_resp(stack->ctx, req, NEU_ERR_PLUGIN_DISCONNECTED);
-        plog_warn((neu_plugin_t *) stack->ctx, "send write req fail, %hhu!%hu",
-                  slave_id, start_address);
+        if (response) {
+            stack->write_resp(stack->ctx, req, NEU_ERR_PLUGIN_DISCONNECTED);
+            plog_warn((neu_plugin_t *) stack->ctx,
+                      "send write req fail, %hhu!%hu", slave_id, start_address);
+        }
     }
     return ret;
 }

plugins/modbus/modbus_stack.h

@@ -52,7 +52,7 @@ int  modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
 int  modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,
                         enum modbus_area area, uint16_t start_address,
                         uint16_t n_reg, uint8_t *bytes, uint8_t n_byte,
-                        uint16_t *response_size);
+                        uint16_t *response_size, bool response);
 bool modbus_stack_is_rtu(modbus_stack_t *stack);
 
 #endif

plugins/modbus/modbus_tcp.c

@@ -29,7 +29,7 @@
 static neu_plugin_t *driver_open(void);
 
 static int driver_close(neu_plugin_t *plugin);
-static int driver_init(neu_plugin_t *plugin);
+static int driver_init(neu_plugin_t *plugin, bool load);
 static int driver_uninit(neu_plugin_t *plugin);
 static int driver_start(neu_plugin_t *plugin);
 static int driver_stop(neu_plugin_t *plugin);
@@ -42,6 +42,7 @@ static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
 static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
 static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
                         neu_value_u value);
+static int driver_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags);
 
 static const neu_plugin_intf_funs_t plugin_intf_funs = {
     .open    = driver_open,
@@ -57,7 +58,10 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
     .driver.group_timer   = driver_group_timer,
     .driver.write_tag     = driver_write,
     .driver.tag_validator = driver_tag_validator,
-    .driver.write_tags    = NULL,
+    .driver.write_tags    = driver_write_tags,
+    .driver.add_tags      = NULL,
+    .driver.load_tags     = NULL,
+    .driver.del_tags      = NULL,
 };
 
 const neu_plugin_module_t neu_plugin_module = {
@@ -94,8 +98,9 @@ static int driver_close(neu_plugin_t *plugin)
     return 0;
 }
 
-static int driver_init(neu_plugin_t *plugin)
+static int driver_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     plugin->protocol = MODBUS_PROTOCOL_TCP;
     plugin->events   = neu_event_new();
     plugin->stack    = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_TCP,
@@ -144,7 +149,9 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     char *           err_param = NULL;
     neu_json_elem_t  port      = { .name = "port", .t = NEU_JSON_INT };
     neu_json_elem_t  timeout   = { .name = "timeout", .t = NEU_JSON_INT };
-    neu_json_elem_t  host      = { .name = "host", .t = NEU_JSON_STR };
+    neu_json_elem_t  host      = { .name      = "host",
+                             .t         = NEU_JSON_STR,
+                             .v.val_str = NULL };
     neu_json_elem_t  interval  = { .name = "interval", .t = NEU_JSON_INT };
     neu_json_elem_t  mode  = { .name = "connection_mode", .t = NEU_JSON_INT };
     neu_json_elem_t  tmode = { .name = "transport_mode", .t = NEU_JSON_INT };
@@ -159,7 +166,9 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     if (ret != 0) {
         plog_error(plugin, "config: %s, decode error: %s", config, err_param);
         free(err_param);
-        free(host.v.val_str);
+        if (host.v.val_str != NULL) {
+            free(host.v.val_str);
+        }
         return -1;
     }
 
@@ -275,5 +284,10 @@ static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
 static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
                         neu_value_u value)
 {
-    return modbus_write(plugin, req, tag, value);
+    return modbus_write_tag(plugin, req, tag, value);
+}
+
+static int driver_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
+{
+    return modbus_write_tags(plugin, req, tags);
 }

plugins/monitor/monitor.c

@@ -50,9 +50,8 @@ static void disconnect_cb(void *data)
 
 static nng_http_server *server_init()
 {
-    nng_url *        url       = NULL;
-    nng_http_server *server    = NULL;
-    const char *     host_port = "http://0.0.0.0:7000";
+    nng_url *        url    = NULL;
+    nng_http_server *server = NULL;
 
     nlog_notice("monitor bind url: %s", host_port);
 
@@ -292,10 +291,11 @@ static int monitor_plugin_close(neu_plugin_t *plugin)
     return rv;
 }
 
-static int monitor_plugin_init(neu_plugin_t *plugin)
+static int monitor_plugin_init(neu_plugin_t *plugin, bool load)
 {
     int rv = 0;
     (void) plugin;
+    (void) load;
 
     plog_notice(plugin, "Initialize plugin: %s", neu_plugin_module.module_name);
     return rv;

plugins/mqtt/mqtt_plugin.c

@@ -57,8 +57,9 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
     return NEU_ERR_SUCCESS;
 }
 
-static int mqtt_plugin_init(neu_plugin_t *plugin)
+static int mqtt_plugin_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     plog_notice(plugin, "initialize plugin `%s` success",
                 neu_plugin_module.module_name);
     neu_adapter_register_metric_cb_t register_metric =

plugins/restful/rest.c

@@ -52,10 +52,8 @@ struct neu_plugin {
 static nng_http_server *server_init()
 {
     nng_url *        url;
-    char             host_port[24] = { 0 };
     nng_http_server *server;
 
-    snprintf(host_port, sizeof(host_port), "http://0.0.0.0:7000");
     nlog_notice("bind url: %s", host_port);
 
     int ret = nng_url_parse(&url, host_port);
@@ -130,8 +128,9 @@ static int dashb_plugin_close(neu_plugin_t *plugin)
     return rv;
 }
 
-static int dashb_plugin_init(neu_plugin_t *plugin)
+static int dashb_plugin_init(neu_plugin_t *plugin, bool load)
 {
+    (void) load;
     int rv = 0;
 
     (void) plugin;

src/adapter/adapter.c

@@ -65,6 +65,8 @@ static const adapter_callbacks_t callback_funs = {
     .update_metric   = adapter_update_metric,
 };
 
+static __thread int create_adapter_error = 0;
+
 #define REGISTER_METRIC(adapter, name, init) \
     adapter_register_metric(adapter, name, name##_HELP, name##_TYPE, init);
 
@@ -108,9 +110,20 @@ static inline void stop_log_level_timer(neu_adapter_t *adapter)
     adapter->timer_lev = NULL;
 }
 
-neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
+int neu_adapter_error()
+{
+    return create_adapter_error;
+}
+
+void neu_adapter_set_error(int error)
+{
+    create_adapter_error = error;
+}
+
+neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
 {
     int                  rv      = 0;
+    int                  init_rv = 0;
     neu_adapter_t *      adapter = NULL;
     neu_event_io_param_t param   = { 0 };
 
@@ -170,7 +183,8 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
 
     zlog_level_switch(common->log, default_log_level);
 
-    adapter->module->intf_funs->init(adapter->plugin);
+    init_rv = adapter->module->intf_funs->init(adapter->plugin, load);
+
     if (adapter_load_setting(adapter->name, &adapter->setting) == 0) {
         if (adapter->module->intf_funs->setting(adapter->plugin,
                                                 adapter->setting) == 0) {
@@ -198,7 +212,23 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     nlog_notice("Success to create adapter: %s", adapter->name);
 
     adapter_storage_state(adapter->name, adapter->state);
-    return adapter;
+
+    if (init_rv != 0) {
+        nlog_warn("Failed to init adapter: %s", adapter->name);
+        neu_adapter_set_error(init_rv);
+
+        if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
+            neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter);
+        }
+        stop_log_level_timer(adapter);
+        neu_event_del_io(adapter->events, adapter->nng_io);
+
+        neu_adapter_destroy(adapter);
+        return NULL;
+    } else {
+        neu_adapter_set_error(0);
+        return adapter;
+    }
 }
 
 int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
@@ -782,9 +812,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
             for (int i = 0; i < cmd->n_tag; i++) {
-                int ret =
-                    neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter,
-                                               cmd->group, &cmd->tags[i]);
+                int ret = neu_adapter_driver_validate_tag(
+                    (neu_adapter_driver_t *) adapter, cmd->group,
+                    &cmd->tags[i]);
                 if (ret == 0) {
                     resp.index += 1;
                 } else {
@@ -796,6 +826,28 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
             resp.error = NEU_ERR_GROUP_NOT_ALLOW;
         }
 
+        if (resp.index > 0) {
+            int ret = neu_adapter_driver_try_add_tag(
+                (neu_adapter_driver_t *) adapter, cmd->group, cmd->tags,
+                resp.index);
+            if (ret != 0) {
+                resp.index = 0;
+                resp.error = ret;
+            }
+        }
+
+        for (int i = 0; i < resp.index; i++) {
+            int ret = neu_adapter_driver_add_tag(
+                (neu_adapter_driver_t *) adapter, cmd->group, &cmd->tags[i]);
+            if (ret != 0) {
+                neu_adapter_driver_try_del_tag((neu_adapter_driver_t *) adapter,
+                                               resp.index - i);
+                resp.index = i;
+                resp.error = ret;
+                break;
+            }
+        }
+
         for (uint16_t i = resp.index; i < cmd->n_tag; i++) {
             neu_tag_fini(&cmd->tags[i]);
         }

src/adapter/adapter_internal.h

@@ -63,7 +63,10 @@ typedef struct adapter_msg_handler {
     adapter_handler    handler;
 } adapter_msg_handler_t;
 
-neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
+int  neu_adapter_error();
+void neu_adapter_set_error(int error);
+
+neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load);
 void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state);
 
 int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name);

src/adapter/driver/driver.c

@@ -113,6 +113,10 @@ static void update_im(neu_adapter_t *adapter, const char *group,
     }
 
     UT_array *tags = neu_adapter_driver_get_ptag(driver, group, tag);
+    if (tags == NULL) {
+        return;
+    }
+
     neu_driver_cache_update_change(driver->cache, group, tag, global_timestamp,
                                    value, true);
     driver->adapter.cb_funs.update_metric(&driver->adapter,
@@ -242,6 +246,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
     {
         HASH_DEL(driver->groups, el);
 
+        neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(el->group));
         neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
         neu_event_del_timer(driver->driver_events, el->read);
         neu_event_del_timer(driver->driver_events, el->write);
@@ -379,7 +384,7 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
         switch (tag->option.value32.endian) {
         case NEU_DATATAG_ENDIAN_LB32:
             neu_ntohs_p((uint16_t *) value->value.bytes);
-            neu_ntohs_p((uint16_t *) value->value.bytes + 2);
+            neu_ntohs_p((uint16_t *) (value->value.bytes + 2));
             break;
         case NEU_DATATAG_ENDIAN_BB32:
             value->value.u32 = htonl(value->value.u32);
@@ -387,7 +392,7 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
         case NEU_DATATAG_ENDIAN_BL32:
             value->value.u32 = htonl(value->value.u32);
             neu_ntohs_p((uint16_t *) value->value.bytes);
-            neu_ntohs_p((uint16_t *) value->value.bytes + 2);
+            neu_ntohs_p((uint16_t *) (value->value.bytes + 2));
             break;
         case NEU_DATATAG_ENDIAN_LL32:
         default:
@@ -403,7 +408,7 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
         switch (tag->option.value32.endian) {
         case NEU_DATATAG_ENDIAN_LB32:
             neu_ntohs_p((uint16_t *) value->value.bytes);
-            neu_ntohs_p((uint16_t *) value->value.bytes + 2);
+            neu_ntohs_p((uint16_t *) (value->value.bytes + 2));
             break;
         case NEU_DATATAG_ENDIAN_BB32:
             value->value.u32 = htonl(value->value.u32);
@@ -411,7 +416,7 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
         case NEU_DATATAG_ENDIAN_BL32:
             value->value.u32 = htonl(value->value.u32);
             neu_ntohs_p((uint16_t *) value->value.bytes);
-            neu_ntohs_p((uint16_t *) value->value.bytes + 2);
+            neu_ntohs_p((uint16_t *) (value->value.bytes + 2));
             break;
         case NEU_DATATAG_ENDIAN_LL32:
         default:
@@ -733,6 +738,8 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
     if (find != NULL) {
         HASH_DEL(driver->groups, find);
 
+        neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(find->group));
+
         neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
         neu_event_del_timer(driver->driver_events, find->read);
         neu_event_del_timer(driver->driver_events, find->write);
@@ -769,9 +776,9 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
         utarray_free(find->grp.tags);
         utarray_free(find->wt_tags);
         neu_group_destroy(find->group);
+        pthread_mutex_destroy(&find->wt_mtx);
         free(find);
 
-        pthread_mutex_destroy(&find->wt_mtx);
         neu_adapter_del_group_metrics(&driver->adapter, name);
         ret = NEU_ERR_SUCCESS;
     }
@@ -823,11 +830,43 @@ UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver)
     return groups;
 }
 
-int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
-                               neu_datatag_t *tag)
+int neu_adapter_driver_try_add_tag(neu_adapter_driver_t *driver,
+                                   const char *group, neu_datatag_t *tags,
+                                   int n_tag)
 {
-    int      ret  = NEU_ERR_SUCCESS;
-    group_t *find = NULL;
+    int ret = 0;
+    if (driver->adapter.module->intf_funs->driver.add_tags != NULL) {
+        ret = driver->adapter.module->intf_funs->driver.add_tags(
+            driver->adapter.plugin, group, tags, n_tag);
+    }
+    return ret;
+}
+
+int neu_adapter_driver_load_tag(neu_adapter_driver_t *driver, const char *group,
+                                neu_datatag_t *tags, int n_tag)
+{
+    int ret = 0;
+    if (driver->adapter.module->intf_funs->driver.load_tags != NULL) {
+        ret = driver->adapter.module->intf_funs->driver.load_tags(
+            driver->adapter.plugin, group, tags, n_tag);
+    }
+    return ret;
+}
+
+int neu_adapter_driver_try_del_tag(neu_adapter_driver_t *driver, int n_tag)
+{
+    int ret = 0;
+    if (driver->adapter.module->intf_funs->driver.del_tags != NULL) {
+        ret = driver->adapter.module->intf_funs->driver.del_tags(
+            driver->adapter.plugin, n_tag);
+    }
+    return ret;
+}
+
+int neu_adapter_driver_validate_tag(neu_adapter_driver_t *driver,
+                                    const char *group, neu_datatag_t *tag)
+{
+    (void) group;
 
     if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
@@ -851,7 +890,7 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
     }
 
     if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) {
-        ret = driver->adapter.module->intf_funs->driver.validate_tag(
+        int ret = driver->adapter.module->intf_funs->driver.validate_tag(
             driver->adapter.plugin, tag);
         if (ret != NEU_ERR_SUCCESS) {
             return ret;
@@ -860,6 +899,17 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
 
     neu_datatag_parse_addr_option(tag, &tag->option);
 
+    return NEU_ERR_SUCCESS;
+}
+
+int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
+                               neu_datatag_t *tag)
+{
+    int      ret  = NEU_ERR_SUCCESS;
+    group_t *find = NULL;
+
+    neu_datatag_parse_addr_option(tag, &tag->option);
+
     HASH_FIND_STR(driver->groups, group, find);
     if (find == NULL) {
         neu_adapter_driver_add_group(driver, group, 3000);
@@ -892,6 +942,7 @@ int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,
     }
 
     if (ret == NEU_ERR_SUCCESS) {
+        neu_adapter_driver_try_del_tag(driver, 1);
         neu_adapter_update_group_metric(&driver->adapter, group,
                                         NEU_METRIC_GROUP_TAGS_TOTAL,
                                         neu_group_tag_size(find->group));
@@ -1186,7 +1237,7 @@ static int read_callback(void *usr_data)
                               group_change);
     }
 
-    if (group->grp.tags != NULL) {
+    if (group->grp.tags != NULL && utarray_len(group->grp.tags) > 0) {
         int64_t spend = global_timestamp;
 
         group->driver->adapter.module->intf_funs->driver.group_timer(

src/adapter/driver/driver_internal.h

@@ -51,6 +51,16 @@ int neu_adapter_driver_group_exist(neu_adapter_driver_t *driver,
                                    const char *          name);
 UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver);
 
+int neu_adapter_driver_try_del_tag(neu_adapter_driver_t *driver, int n_tag);
+int neu_adapter_driver_try_add_tag(neu_adapter_driver_t *driver,
+                                   const char *group, neu_datatag_t *tags,
+                                   int n_tag);
+int neu_adapter_driver_load_tag(neu_adapter_driver_t *driver, const char *group,
+                                neu_datatag_t *tags, int n_tag);
+
+int neu_adapter_driver_validate_tag(neu_adapter_driver_t *driver,
+                                    const char *group, neu_datatag_t *tag);
+
 int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
                                neu_datatag_t *tag);
 int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,

src/adapter/storage.c

@@ -157,6 +157,7 @@ int adapter_load_group_and_tag(neu_adapter_driver_t *driver)
         utarray_foreach(tags, neu_datatag_t *, tag)
         {
             neu_adapter_driver_add_tag(driver, p->name, tag);
+            neu_adapter_driver_load_tag(driver, p->name, tag, 1);
         }
 
         utarray_free(tags);

src/argparse.c

@@ -18,6 +18,7 @@
  **/
 
 #include <errno.h>
+#include <fcntl.h>
 #include <getopt.h>
 #include <inttypes.h>
 #include <stdio.h>
@@ -31,7 +32,10 @@
 
 #include "argparse.h"
 #include "persist/persist.h"
+#include "utils/log.h"
 #include "version.h"
+#include "json/json.h"
+#include "json/neu_json_param.h"
 
 #define OPTIONAL_ARGUMENT_IS_PRESENT                             \
     ((optarg == NULL && optind < argc && argv[optind][0] != '-') \
@@ -60,7 +64,7 @@ const char *usage_text =
 "OPTIONS:\n"
 "    -d, --daemon         run as daemon process\n"
 "    -h, --help           show this help message\n"
-"    --stop               stop running neuron\n"
+"    stop                 stop running neuron\n"
 "    --log                log to the stdout\n"
 "    --log_level <LEVEL>  default log level(DEBUG,NOTICE)\n"
 "    --reset-password     reset dashboard to use default password\n"
@@ -71,6 +75,7 @@ const char *usage_text =
 "                           - NUMBER,     restart max NUMBER of times\n"
 "    --version            print version information\n"
 "    --disable_auth       disable http api auth\n"
+"    --config_file <PATH> startup parameter configuration file\n"
 "    --config_dir <DIR>   directory from which neuron reads configuration\n"
 "    --plugin_dir <DIR>   directory from which neuron loads plugin lib files\n"
 "\n";
@@ -133,6 +138,189 @@ static inline bool file_exists(const char *const path)
     return -1 != stat(path, &buf);
 }
 
+static inline int load_spec_arg(int argc, char *argv[], neu_cli_args_t *args)
+{
+    int ret = 0;
+    if (argc > 1 && strcmp(argv[1], "stop") == 0) {
+        args->stop = true;
+    }
+    return ret;
+}
+
+static inline int load_env(neu_cli_args_t *args, char **log_level_out,
+                           char **config_dir_out, char **plugin_dir_out)
+{
+
+    int   ret    = 0;
+    char *daemon = getenv(NEU_ENV_DAEMON);
+    if (daemon != NULL) {
+        if (strcmp(daemon, "1") == 0) {
+            args->daemonized = true;
+        } else {
+            args->daemonized = false;
+        }
+    }
+
+    char *log = getenv(NEU_ENV_LOG);
+    if (log != NULL) {
+        if (strcmp(log, "1") == 0) {
+            args->dev_log = true;
+        } else {
+            args->dev_log = false;
+        }
+    }
+
+    char *log_level = getenv(NEU_ENV_LOG_LEVEL);
+    if (log_level != NULL) {
+        if (*log_level_out != NULL) {
+            free(*log_level_out);
+        }
+        *log_level_out = strdup(log_level);
+    }
+
+    char *restart = getenv(NEU_ENV_RESTART);
+    if (restart != NULL) {
+        int t = parse_restart_policy(restart, &args->restart);
+        if (t < 0) {
+            printf("neuron NEU_ENV_RESTART setting error!\n");
+            ret = -1;
+        }
+    }
+
+    char *disable_auth = getenv(NEU_ENV_DISABLE_AUTH);
+    if (disable_auth != NULL) {
+        if (strcmp(disable_auth, "1") == 0) {
+            args->disable_auth = true;
+        } else {
+            args->disable_auth = false;
+        }
+    }
+
+    char *config_dir = getenv(NEU_ENV_CONFIG_DIR);
+    if (config_dir != NULL) {
+        if (*config_dir_out != NULL) {
+            free(*config_dir_out);
+        }
+        *config_dir_out = strdup(config_dir);
+    }
+
+    char *plugin_dir = getenv(NEU_ENV_PLUGIN_DIR);
+    if (plugin_dir != NULL) {
+        if (*plugin_dir_out != NULL) {
+            free(*plugin_dir_out);
+        }
+        *plugin_dir_out = strdup(plugin_dir);
+    }
+
+    return ret;
+}
+
+static inline void resolve_config_file_path(int argc, char *argv[],
+                                            struct option *long_options,
+                                            char *opts, char **config_file)
+{
+    int c            = 0;
+    int option_index = 0;
+    while ((c = getopt_long(argc, argv, opts, long_options, &option_index)) !=
+           -1) {
+        switch (c) {
+        case 'c':
+            if (0 == strcmp("config_file", long_options[option_index].name)) {
+                *config_file = strdup(optarg);
+                goto end;
+            }
+            break;
+        default:
+            break;
+        }
+    }
+end:
+    optind = 0;
+}
+
+static inline int load_config_file(int argc, char *argv[],
+                                   struct option *long_options, char *opts,
+                                   neu_cli_args_t *args)
+{
+    char *config_file = NULL;
+    int   ret         = -1;
+    int   fd          = -1;
+    void *root        = NULL;
+
+    resolve_config_file_path(argc, argv, long_options, opts, &config_file);
+
+    do {
+        if (!config_file) {
+            config_file = strdup(NEURON_CONFIG_FNAME);
+        }
+
+        if (!file_exists(config_file)) {
+            printf("configuration file `%s` not exists\n", config_file);
+            nlog_error("configuration file `%s` not exists\n", config_file);
+            break;
+        }
+
+        char buf[512] = { 0 };
+        fd            = open(config_file, O_RDONLY);
+        if (fd < 0) {
+            printf("cannot open %s reason: %s\n", config_file, strerror(errno));
+            break;
+        }
+        int size = read(fd, buf, sizeof(buf) - 1);
+        if (size <= 0) {
+            printf("cannot read %s reason: %s\n", config_file, strerror(errno));
+            break;
+        }
+
+        root = neu_json_decode_new(buf);
+        if (root == NULL) {
+            printf("config file %s foramt error!\n", config_file);
+            nlog_error("config file %s foramt error!\n", config_file);
+            break;
+        }
+
+        neu_json_elem_t elems[] = { { .name = "disable_auth",
+                                      .t    = NEU_JSON_INT },
+                                    { .name = "ip", .t = NEU_JSON_STR },
+                                    { .name = "port", .t = NEU_JSON_INT } };
+
+        if (neu_json_decode_by_json(root, 3, elems) == 0) {
+            if (elems[0].v.val_int != 0) {
+                args->disable_auth = true;
+            } else {
+                args->disable_auth = false;
+            }
+
+            if (elems[1].v.val_str != NULL) {
+                args->ip = strdup(elems[1].v.val_str);
+                free(elems[1].v.val_str);
+            }
+
+            args->port = elems[2].v.val_int;
+            ret        = 0;
+        } else {
+            printf("config file %s elems error! must had ip, port and "
+                   "disable_auth.\n",
+                   config_file);
+        }
+
+    } while (0);
+
+    if (config_file) {
+        free(config_file);
+    }
+
+    if (fd) {
+        close(fd);
+    }
+
+    if (root) {
+        neu_json_decode_free(root);
+    }
+
+    return ret;
+}
+
 void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
 {
     int           ret                 = 0;
@@ -150,6 +338,7 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
         { "restart", required_argument, NULL, 'r' },
         { "version", no_argument, NULL, 'v' },
         { "disable_auth", no_argument, NULL, 'a' },
+        { "config_file", required_argument, NULL, 'c' },
         { "config_dir", required_argument, NULL, 'c' },
         { "plugin_dir", required_argument, NULL, 'p' },
         { "stop", no_argument, NULL, 's' },
@@ -161,6 +350,23 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
     int c            = 0;
     int option_index = 0;
 
+    if (load_spec_arg(argc, argv, args) < 0) {
+        ret = 1;
+        goto quit;
+    }
+
+    // load config file
+    if (load_config_file(argc, argv, long_options, opts, args) < 0) {
+        ret = 1;
+        goto quit;
+    }
+
+    // load env
+    if (load_env(args, &log_level, &config_dir, &plugin_dir) < 0) {
+        ret = 1;
+        goto quit;
+    }
+
     while ((c = getopt_long(argc, argv, opts, long_options, &option_index)) !=
            -1) {
         switch (c) {
@@ -177,6 +383,9 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
             args->dev_log = true;
             break;
         case 'o':
+            if (log_level != NULL) {
+                free(log_level);
+            }
             log_level = strdup(optarg);
             break;
         case 'r':
@@ -197,14 +406,19 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
             args->disable_auth = true;
             break;
         case 'c':
-            config_dir = strdup(optarg);
+            if (0 == strcmp("config_dir", long_options[option_index].name)) {
+                if (config_dir != NULL) {
+                    free(config_dir);
+                }
+                config_dir = strdup(optarg);
+            }
             break;
         case 'p':
+            if (plugin_dir != NULL) {
+                free(plugin_dir);
+            }
             plugin_dir = strdup(optarg);
             break;
-        case 's':
-            args->stop = true;
-            break;
         case '?':
         default:
             usage();
@@ -278,5 +492,6 @@ void neu_cli_args_fini(neu_cli_args_t *args)
         free(args->log_init_file);
         free(args->config_dir);
         free(args->plugin_dir);
+        free(args->ip);
     }
 }

src/argparse.h

@@ -26,6 +26,16 @@
 #define NEU_RESTART_ALWAYS ((size_t) -2)
 #define NEU_RESTART_ONFAILURE ((size_t) -1)
 
+#define NEU_ENV_DAEMON "NEURON_DAEMON"
+#define NEU_ENV_LOG "NEURON_LOG"
+#define NEU_ENV_LOG_LEVEL "NEURON_LOG_LEVEL"
+#define NEU_ENV_RESTART "NEURON_RESTART"
+#define NEU_ENV_DISABLE_AUTH "NEURON_DISABLE_AUTH"
+#define NEU_ENV_CONFIG_DIR "NEURON_CONFIG_DIR"
+#define NEU_ENV_PLUGIN_DIR "NEURON_PLUGIN_DIR"
+
+#define NEURON_CONFIG_FNAME "./config/neuron.json"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -47,6 +57,8 @@ typedef struct {
     char * config_dir;
     char * plugin_dir;
     bool   stop;
+    char * ip;
+    int    port;
 } neu_cli_args_t;
 
 /** Parse command line arguments.

src/connection/connection.c

@@ -1119,15 +1119,13 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
             } else if (used == -1) {
                 neu_conn_disconnect(conn);
                 break;
+            } else {
+                conn->offset -= used;
+                memmove(conn->buf, conn->buf + used, conn->offset);
+                neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
+                                             conn->offset);
             }
         }
-        if (conn->offset != 0) {
-            conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
-            memmove(conn->buf,
-                    conn->buf +
-                        neu_protocol_unpack_buf_used_size(&protocol_buf),
-                    conn->offset);
-        }
     }
 
     return ret;
@@ -1150,15 +1148,13 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
             } else if (used == -1) {
                 neu_conn_tcp_server_close_client(conn, fd);
                 break;
+            } else {
+                conn->offset -= used;
+                memmove(conn->buf, conn->buf + used, conn->offset);
+                neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
+                                             conn->offset);
             }
         }
-        if (conn->offset != 0) {
-            conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
-            memmove(conn->buf,
-                    conn->buf +
-                        neu_protocol_unpack_buf_used_size(&protocol_buf),
-                    conn->offset);
-        }
     }
     return ret;
 }

src/connection/mqtt_client.c

@@ -1025,6 +1025,12 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
         goto end;
     }
 
+    if (NULL == client->host) {
+        log(error, "no client host");
+        rv = -1;
+        goto end;
+    }
+
     if (client->tls_cfg) {
         cfg = client->tls_cfg;
     } else {
@@ -1037,6 +1043,12 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
         client->tls_cfg = cfg;
     }
 
+    if (0 != (rv = nng_tls_config_server_name(client->tls_cfg, client->host))) {
+        log(error, "nng_tls_config_server_name fail: %s", nng_strerror(rv));
+        rv = -1;
+        goto end;
+    }
+
     if (cert != NULL && key != NULL) {
         if ((rv = nng_tls_config_auth_mode(cfg, NNG_TLS_AUTH_MODE_REQUIRED)) !=
             0) {

src/core/manager.c

@@ -564,7 +564,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_ADD_NODE: {
         neu_req_add_node_t *cmd = (neu_req_add_node_t *) &header[1];
         int                 error =
-            neu_manager_add_node(manager, cmd->node, cmd->plugin, false);
+            neu_manager_add_node(manager, cmd->node, cmd->plugin, false, false);
         neu_resp_error_t e = { .error = error };
 
         if (error == NEU_ERR_SUCCESS) {
@@ -1096,7 +1096,7 @@ static void start_static_adapter(neu_manager_t *manager, const char *name)
     adapter_info.handle = instance.handle;
     adapter_info.module = instance.module;
 
-    adapter = neu_adapter_create(&adapter_info);
+    adapter = neu_adapter_create(&adapter_info, true);
     neu_node_manager_add_static(manager->node_manager, adapter);
     neu_adapter_init(adapter, false);
     neu_adapter_start(adapter);
@@ -1119,7 +1119,7 @@ static void start_single_adapter(neu_manager_t *manager, const char *name,
 
     adapter_info.handle = instance.handle;
     adapter_info.module = instance.module;
-    adapter             = neu_adapter_create(&adapter_info);
+    adapter             = neu_adapter_create(&adapter_info, true);
 
     neu_node_manager_add_single(manager->node_manager, adapter, display);
     if (display) {

src/core/manager_internal.c

@@ -19,6 +19,7 @@
 #include <dlfcn.h>
 
 #include "utils/log.h"
+#include "json/neu_json_param.h"
 
 #include "adapter.h"
 #include "errcodes.h"
@@ -46,7 +47,7 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager)
 
 int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
                          const char *             plugin_name,
-                         neu_node_running_state_e state)
+                         neu_node_running_state_e state, bool load)
 {
     neu_adapter_t *       adapter      = NULL;
     neu_plugin_instance_t instance     = { 0 };
@@ -78,7 +79,10 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
     adapter_info.handle = instance.handle;
     adapter_info.module = instance.module;
 
-    adapter = neu_adapter_create(&adapter_info);
+    adapter = neu_adapter_create(&adapter_info, load);
+    if (adapter == NULL) {
+        return neu_adapter_error();
+    }
     neu_node_manager_add(manager->node_manager, adapter);
     neu_adapter_init(adapter, state);
 
@@ -517,7 +521,7 @@ int neu_manager_instantiate_template(neu_manager_t *          manager,
     }
 
     int ret = neu_manager_add_node(manager, req->node,
-                                   neu_template_plugin(tmpl), false);
+                                   neu_template_plugin(tmpl), false, false);
     if (0 != ret) {
         return ret;
     }
@@ -617,6 +621,19 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
         return NEU_ERR_NODE_NOT_EXIST;
     }
 
+    // guard against empty mqtt topic parameter
+    // this is not an elegant solution due to the current architecture
+    if (params && 0 == strcmp(adapter->module->module_name, "MQTT")) {
+        neu_json_elem_t elem = { .name = "topic", .t = NEU_JSON_STR };
+        neu_json_decode((char *) params, 1, &elem);
+        neu_parse_param(params, NULL, 1, &elem);
+        if (elem.v.val_str && 0 == strlen(elem.v.val_str)) {
+            free(elem.v.val_str);
+            return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+        }
+        free(elem.v.val_str);
+    }
+
     if (NEU_NA_TYPE_APP != neu_adapter_get_type(adapter)) {
         return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE;
     }

src/core/manager_internal.h

@@ -53,7 +53,7 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager);
 
 int       neu_manager_add_node(neu_manager_t *manager, const char *node_name,
                                const char *             plugin_name,
-                               neu_node_running_state_e state);
+                               neu_node_running_state_e state, bool load);
 int       neu_manager_del_node(neu_manager_t *manager, const char *node_name);
 UT_array *neu_manager_get_nodes(neu_manager_t *manager, int type,
                                 const char *plugin, const char *node);

src/core/storage.c

@@ -363,7 +363,8 @@ int manager_load_node(neu_manager_t *manager)
     utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
     {
         rv                    = neu_manager_add_node(manager, node_info->name,
-                                  node_info->plugin_name, node_info->state);
+                                  node_info->plugin_name, node_info->state,
+                                  true);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
         nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",
                     ok_or_err, node_info->type, node_info->name,

src/daemon.c

@@ -31,9 +31,13 @@
 #include <string.h>
 
 #include "utils/log.h"
+#include "utils/time.h"
 
 #include "daemon.h"
 
+#define STOP_TIMEOUT 20000
+#define STOP_TICK 2000
+
 void daemonize()
 {
     int              nfile;
@@ -95,6 +99,45 @@ static inline int lock_file(int fd)
     return fcntl(fd, F_SETLK, &fl);
 }
 
+static int neuron_wait_exit(uint32_t ms)
+{
+    uint32_t tick = 0;
+    int      ret  = -1;
+    int      fd   = -1;
+
+    fd = open(NEURON_DAEMON_LOCK_FNAME, O_WRONLY);
+    if (fd < 0) {
+        nlog_error("cannot open %s reason: %s\n", NEURON_DAEMON_LOCK_FNAME,
+                   strerror(errno));
+        exit(1);
+    }
+
+    do {
+        // try to lock file
+        if (lock_file(fd) < 0) {
+            if (EACCES == errno || EAGAIN == errno) {
+                // a neuron process is running
+                printf("neuron is stopping.\n");
+            } else {
+                close(fd);
+                nlog_error("cannot lock %s reason: %s\n",
+                           NEURON_DAEMON_LOCK_FNAME, strerror(errno));
+                exit(1);
+            }
+        } else {
+            printf("neuron had stopped.\n");
+            ret = 0;
+            break;
+        }
+
+        neu_msleep(STOP_TICK);
+        tick += STOP_TICK;
+
+    } while (tick < ms);
+    close(fd);
+    return ret;
+}
+
 int neuron_already_running()
 {
     int     fd      = -1;
@@ -160,7 +203,9 @@ int neuron_stop()
     }
 
     if (0 == kill((pid_t)(-gid), SIGINT)) {
-        ret = 0;
+        if (neuron_wait_exit(STOP_TIMEOUT) == 0) {
+            ret = 0;
+        }
     } else {
         nlog_error("cannot kill gpid:%ld reason: %s\n", (long) gid,
                    strerror(errno));

src/event/event_linux.c

@@ -260,6 +260,10 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
 
 int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
 {
+    if (io == NULL) {
+        return 0;
+    }
+
     struct event_data *data = (struct event_data *) io->event_data;
     zlog_notice(neuron, "del io: %d from epoll: %d", io->fd, events->epoll_fd);
 

src/main.c

@@ -35,6 +35,7 @@ static neu_manager_t *g_manager         = NULL;
 zlog_category_t *     neuron            = NULL;
 bool                  disable_jwt       = false;
 int                   default_log_level = ZLOG_LEVEL_NOTICE;
+char                  host_port[24]     = { 0 };
 
 int64_t global_timestamp = 0;
 
@@ -98,6 +99,7 @@ int main(int argc, char *argv[])
     neu_cli_args_init(&args, argc, argv);
 
     disable_jwt = args.disable_auth;
+    snprintf(host_port, sizeof(host_port), "http://%s:%d", args.ip, args.port);
 
     if (args.daemonized) {
         // become a daemon, this should be before calling `init`
@@ -115,6 +117,8 @@ int main(int argc, char *argv[])
             nlog_notice("neuron stop ret=%d", rv);
             if (rv == 0) {
                 printf("neuron stop successfully.\n");
+            } else {
+                printf("neuron stop failed.\n");
             }
         } else {
             nlog_fatal("neuron process already running, exit.");

tests/README.md

@@ -16,7 +16,7 @@ Generate image files .gcno and data files .gcda under build/tests/CMakeFiles/.di
 Execute the following command in the build folder to generate an intermediate file for HTML:
 
 ```shell
-$ lcov --capture --directory . -- output-file testneuron.info --test-name testneuron
+$ lcov --capture --directory . --output-file testneuron.info --test-name testneuron
 ```
 
 Execute the following command in the build folder to generate an HTML file in the result folder under the build folder:

version

@@ -1 +1 @@
-2.5.3
+2.5.4