Merge pull request #1904 from hxy7yx/v2.5-1

plugin(modbus): fix modbus rtu slave check

.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,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.5.2
-  UI_VERSION: 2.5.2
+  VERSION: 2.5.5
+  UI_VERSION: 2.5.3
   BRANCH: v2.5
 
 jobs:

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)

default_plugins.json

@@ -4,7 +4,6 @@
 		"libplugin-ekuiper.so",
 		"libplugin-monitor.so",
 		"libplugin-modbus-tcp.so",
-		"libplugin-modbus-rtu.so",
-		"libplugin-modbus-qh-tcp.so"
+		"libplugin-modbus-rtu.so"
 	]
 }

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/adapter.h

@@ -257,8 +257,8 @@ typedef struct neu_resp_error {
 } neu_resp_error_t;
 
 typedef struct {
-    char node[NEU_NODE_NAME_LEN];
-    bool auto_start;
+    char                     node[NEU_NODE_NAME_LEN];
+    neu_node_running_state_e state;
 } neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t;
 
 typedef struct neu_req_add_plugin {

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 \

persistence/0042_2.5.3_fx_rename.sql

@@ -0,0 +1,26 @@
+/**
+ * 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;
+
+UPDATE nodes
+SET plugin_name = 'Mitsubishi FX'
+WHERE plugin_name = 'Mitsabishi FX';
+
+COMMIT;

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/CMakeLists.txt

@@ -5,7 +5,6 @@ set(MODBUS_SRC modbus.c modbus_point.c modbus_req.c modbus_stack.c)
 set(CMAKE_BUILD_RPATH ./)
 file(COPY ${CMAKE_SOURCE_DIR}/plugins/modbus/modbus-tcp.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
 file(COPY ${CMAKE_SOURCE_DIR}/plugins/modbus/modbus-rtu.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
-file(COPY ${CMAKE_SOURCE_DIR}/plugins/modbus/modbus-qh-tcp.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
 # modbus rtu plugin
 set(MODBUS_RTU_PLUGIN plugin-modbus-rtu)
 set(MODBUS_RTU_PLUGIN_SOURCES  modbus_rtu.c
@@ -24,12 +23,3 @@ target_include_directories(${MODBUS_TCP_PLUS_PLUGIN} PRIVATE ${CMAKE_SOURCE_DIR}
 target_sources(${MODBUS_TCP_PLUS_PLUGIN} PRIVATE ${MODBUS_TCP_PLUS_PLUGIN_SOURCES})
 target_link_libraries(${MODBUS_TCP_PLUS_PLUGIN} neuron-base)
 
-# modbus qh tcp plugin
-set(MODBUS_TCP_QH_PLUGIN plugin-modbus-qh-tcp)
-set(MODBUS_TCP_QH_PLUGIN_SOURCES  modbus_qh_tcp.c
-                                    ${MODBUS_SRC})
-add_library(${MODBUS_TCP_QH_PLUGIN} SHARED)
-target_include_directories(${MODBUS_TCP_QH_PLUGIN} PRIVATE ${CMAKE_SOURCE_DIR}/include/neuron)
-target_sources(${MODBUS_TCP_QH_PLUGIN} PRIVATE ${MODBUS_TCP_QH_PLUGIN_SOURCES})
-target_link_libraries(${MODBUS_TCP_QH_PLUGIN} neuron-base)
-

plugins/modbus/modbus-qh-tcp.json

@@ -1,104 +0,0 @@
-{
-	"tag_regex": [
-		{
-			"type": 3,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
-		},
-		{
-			"type": 4,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
-		},
-		{
-			"type": 5,
-			"regex": "^[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
-		},
-		{
-			"type": 6,
-			"regex": "^[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
-		},
-		{
-			"type": 7,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
-		},
-		{
-			"type": 8,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
-		},
-		{
-			"type": 9,
-			"regex": "^[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
-		},
-		{
-			"type": 10,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
-		},
-		{
-			"type": 11,
-			"regex": "^[0-9]+!([0-1][0-9]+|[3-4][0-9]+\\.([0-9]|[0-1][0-5]))$"
-		},
-		{
-			"type": 13,
-			"regex": "^[0-9]+![3-4][0-9]+\\.[0-9]+(H|L|)$"
-		}
-	],
-	"group_interval": 100,
-	"host": {
-		"name": "IP Address",
-		"name_zh": "IP地址",
-		"description": "Local IP in server mode, remote device IP in client mode",
-		"description_zh": "服务端模式中填写本地 IP，客户端模式中填写目标设备 IP",
-		"attribute": "required",
-		"type": "string",
-		"valid": {
-			"regex": "/^((2[0-4]\\d|25[0-5]|[01]?\\d\\d?)\\.){3}(2[0-4]\\d|25[0-5]|[01]?\\d\\d?)$/",
-			"length": 30
-		}
-	},
-	"port": {
-		"name": "Port",
-		"name_zh": "端口号",
-		"description": "Local port in server mode, remote device port in client mode",
-		"description_zh": "服务端模式中填写本地端口号，客户端模式中填写远程设备端口号",
-		"attribute": "required",
-		"type": "int",
-		"default": 502,
-		"valid": {
-			"min": 1,
-			"max": 65535
-		}
-	},
-	"timeout": {
-		"name": "Connection Timeout",
-		"name_zh": "连接超时时间",
-		"description": "Connection timeout(ms)",
-		"description_zh": "连接超时时间，单位为毫秒",
-		"attribute": "required",
-		"type": "int",
-		"default": 3000,
-		"valid": {
-			"min": 1000,
-			"max": 65535
-		}
-	},
-	"connection_mode": {
-		"name": "Connection Mode",
-		"name_zh": "连接模式",
-		"description": "Neuron as the client, or as the server",
-		"description_zh": "Neuron 作为客户端或服务端",
-		"attribute": "required",
-		"type": "map",
-		"default": 0,
-		"valid": {
-			"map": [
-				{
-					"key": "Client",
-					"value": 0
-				},
-				{
-					"key": "Server",
-					"value": 1
-				}
-			]
-		}
-	}
-}

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.h

@@ -40,14 +40,23 @@ typedef enum modbus_action {
 } modbus_action_e;
 
 typedef enum modbus_function {
-    MODBUS_READ_COIL        = 0x1,
-    MODBUS_READ_INPUT       = 0x02,
-    MODBUS_READ_HOLD_REG    = 0x03,
-    MODBUS_READ_INPUT_REG   = 0x04,
-    MODBUS_WRITE_S_COIL     = 0x05,
-    MODBUS_WRITE_S_HOLD_REG = 0x06,
-    MODBUS_WRITE_M_HOLD_REG = 0x10,
-    MODBUS_WRITE_M_COIL     = 0x0F
+    MODBUS_READ_COIL            = 0x1,
+    MODBUS_READ_INPUT           = 0x02,
+    MODBUS_READ_HOLD_REG        = 0x03,
+    MODBUS_READ_INPUT_REG       = 0x04,
+    MODBUS_WRITE_S_COIL         = 0x05,
+    MODBUS_WRITE_S_HOLD_REG     = 0x06,
+    MODBUS_WRITE_M_HOLD_REG     = 0x10,
+    MODBUS_WRITE_M_COIL         = 0x0F,
+    MODBUS_READ_COIL_ERR        = 0x81,
+    MODBUS_READ_INPUT_ERR       = 0x82,
+    MODBUS_READ_HOLD_REG_ERR    = 0x83,
+    MODBUS_READ_INPUT_REG_ERR   = 0x84,
+    MODBUS_WRITE_S_COIL_ERR     = 0x85,
+    MODBUS_WRITE_S_HOLD_REG_ERR = 0x86,
+    MODBUS_WRITE_M_HOLD_REG_ERR = 0x90,
+    MODBUS_WRITE_M_COIL_ERR     = 0x8F,
+    MODBUS_DEVICE_ERR           = -2
 } modbus_function_e;
 
 typedef enum modbus_area {

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_qh_tcp.c

@@ -1,245 +0,0 @@
-/**
- * NEURON IIoT System for Industry 4.0
- * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
- *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 3 of the License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public License
- * along with this program; if not, write to the Free Software Foundation,
- * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
- **/
-#include <stdlib.h>
-
-#include <neuron.h>
-
-#include "errcodes.h"
-
-#include "modbus_point.h"
-#include "modbus_req.h"
-#include "modbus_stack.h"
-
-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_uninit(neu_plugin_t *plugin);
-static int driver_start(neu_plugin_t *plugin);
-static int driver_stop(neu_plugin_t *plugin);
-static int driver_config(neu_plugin_t *plugin, const char *config);
-static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
-                          void *data);
-
-static int driver_tag_validator(const neu_datatag_t *tag);
-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 const neu_plugin_intf_funs_t plugin_intf_funs = {
-    .open    = driver_open,
-    .close   = driver_close,
-    .init    = driver_init,
-    .uninit  = driver_uninit,
-    .start   = driver_start,
-    .stop    = driver_stop,
-    .setting = driver_config,
-    .request = driver_request,
-
-    .driver.validate_tag  = driver_validate_tag,
-    .driver.group_timer   = driver_group_timer,
-    .driver.write_tag     = driver_write,
-    .driver.tag_validator = driver_tag_validator,
-    .driver.write_tags    = NULL,
-};
-
-const neu_plugin_module_t neu_plugin_module = {
-    .version     = NEURON_PLUGIN_VER_1_0,
-    .schema      = "modbus-qh-tcp",
-    .module_name = "Modbus TCP QH",
-    .module_descr =
-        "The modbus-qh-tcp plugin is an optimized version of modbus tcp, which "
-        "expands the packet description field to two bytes, limiting the "
-        "maximum length of Modbus tcp packets to 64KB",
-    .module_descr_zh =
-        "该插件是 modbus-tcp 插件的优化版，数据包描述字段扩展为两个字节,限制 "
-        "Modbus TCP 数据包的最大长度为 64KB。",
-    .intf_funs  = &plugin_intf_funs,
-    .kind       = NEU_PLUGIN_KIND_SYSTEM,
-    .type       = NEU_NA_TYPE_DRIVER,
-    .display    = true,
-    .single     = false,
-    .timer_type = NEU_EVENT_TIMER_NOBLOCK,
-};
-
-static neu_plugin_t *driver_open(void)
-{
-    neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t));
-
-    neu_plugin_common_init(&plugin->common);
-
-    return plugin;
-}
-
-static int driver_close(neu_plugin_t *plugin)
-{
-    free(plugin);
-
-    return 0;
-}
-
-static int driver_init(neu_plugin_t *plugin)
-{
-    plugin->protocol = MODBUS_PROTOCOL_TCP;
-    plugin->events   = neu_event_new();
-    plugin->stack    = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_TCP,
-                                        modbus_send_msg, modbus_value_handle,
-                                        modbus_write_resp);
-
-    plog_notice(plugin, "%s init success", plugin->common.name);
-    return 0;
-}
-
-static int driver_uninit(neu_plugin_t *plugin)
-{
-    plog_notice(plugin, "%s uninit start", plugin->common.name);
-    if (plugin->conn != NULL) {
-        neu_conn_destory(plugin->conn);
-    }
-
-    if (plugin->stack) {
-        modbus_stack_destroy(plugin->stack);
-    }
-
-    neu_event_close(plugin->events);
-
-    plog_notice(plugin, "%s uninit success", plugin->common.name);
-
-    return 0;
-}
-
-static int driver_start(neu_plugin_t *plugin)
-{
-    neu_conn_start(plugin->conn);
-    plog_notice(plugin, "%s start success", plugin->common.name);
-    return 0;
-}
-
-static int driver_stop(neu_plugin_t *plugin)
-{
-    neu_conn_stop(plugin->conn);
-    plog_notice(plugin, "%s stop success", plugin->common.name);
-    return 0;
-}
-
-static int driver_config(neu_plugin_t *plugin, const char *config)
-{
-    int              ret       = 0;
-    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  mode  = { .name = "connection_mode", .t = NEU_JSON_INT };
-    neu_conn_param_t param = { 0 };
-
-    ret = neu_parse_param((char *) config, &err_param, 4, &port, &host, &mode,
-                          &timeout);
-
-    if (ret != 0) {
-        plog_error(plugin, "config: %s, decode error: %s", config, err_param);
-        free(err_param);
-        return -1;
-    }
-
-    param.log = plugin->common.log;
-    if (mode.v.val_int == 1) {
-        param.type                           = NEU_CONN_TCP_SERVER;
-        param.params.tcp_server.ip           = host.v.val_str;
-        param.params.tcp_server.port         = port.v.val_int;
-        param.params.tcp_server.start_listen = modbus_tcp_server_listen;
-        param.params.tcp_server.stop_listen  = modbus_tcp_server_stop;
-        param.params.tcp_server.timeout      = timeout.v.val_int;
-        param.params.tcp_server.max_link     = 1;
-        plugin->is_server                    = true;
-    } else {
-        param.type                      = NEU_CONN_TCP_CLIENT;
-        param.params.tcp_client.ip      = host.v.val_str;
-        param.params.tcp_client.port    = port.v.val_int;
-        param.params.tcp_client.timeout = timeout.v.val_int;
-        plugin->is_server               = false;
-    }
-    plog_notice(plugin,
-                "config: host: %s, port: %" PRId64 ", mode: %" PRId64 "",
-                host.v.val_str, port.v.val_int, mode.v.val_int);
-
-    if (plugin->conn != NULL) {
-        plugin->conn = neu_conn_reconfig(plugin->conn, &param);
-    } else {
-        plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
-        plugin->conn =
-            neu_conn_new(&param, (void *) plugin, modbus_conn_connected,
-                         modbus_conn_disconnected);
-    }
-
-    free(host.v.val_str);
-    return 0;
-}
-
-static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
-                          void *data)
-{
-    (void) plugin;
-    (void) head;
-    (void) data;
-    return 0;
-}
-
-static int driver_tag_validator(const neu_datatag_t *tag)
-{
-    modbus_point_t point = { 0 };
-    return modbus_tag_to_point(tag, &point);
-}
-
-static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
-{
-    modbus_point_t point = { 0 };
-
-    int ret = modbus_tag_to_point(tag, &point);
-    if (ret == 0) {
-        plog_notice(
-            plugin,
-            "validate tag success, name: %s, address: %s, type: %d, slave id: "
-            "%d, start address: %d, n register: %d, area: %s",
-            tag->name, tag->address, tag->type, point.slave_id,
-            point.start_address, point.n_register,
-            modbus_area_to_str(point.area));
-    } else {
-        plog_error(
-            plugin,
-            "validate tag error, name: %s, address: %s, type: %d, slave id: "
-            "%d, start address: %d, n register: %d, area: %s",
-            tag->name, tag->address, tag->type, point.slave_id,
-            point.start_address, point.n_register,
-            modbus_area_to_str(point.area));
-    }
-
-    return ret;
-}
-
-static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
-{
-    return modbus_group_timer(plugin, group, 0xffff);
-}
-
-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);
-}

plugins/modbus/modbus_req.c

@@ -29,8 +29,14 @@ 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, uint16_t response_size);
+static int  process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
+                                 uint16_t response_size);
 
 void modbus_conn_connected(void *data, int fd)
 {
@@ -100,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 {
@@ -162,29 +168,137 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
         plugin->cmd_idx        = i;
         uint16_t response_size = 0;
         uint64_t read_tms      = neu_time_ms();
-
-        int ret = modbus_stack_read(
+        int      ret_buf       = 0;
+        int      ret_r         = modbus_stack_read(
             plugin->stack, gd->cmd_sort->cmd[i].slave_id,
             gd->cmd_sort->cmd[i].area, gd->cmd_sort->cmd[i].start_address,
             gd->cmd_sort->cmd[i].n_register, &response_size);
-        if (ret > 0) {
-            ret = process_protocol_buf(plugin, response_size);
-            rtt = neu_time_ms() - read_tms;
-        }
-        if (ret <= 0) {
+        if (ret_r > 0) {
+            ret_buf = process_protocol_buf(
+                plugin, gd->cmd_sort->cmd[i].slave_id, response_size);
+            if (ret_buf > 0) {
+                rtt = neu_time_ms() - read_tms;
+            } else if (ret_buf == 0) {
+                for (uint16_t j = 0; j < plugin->max_retries; j++) {
+                    ret_r = modbus_stack_read_retry(plugin, gd, i, j,
+                                                    &response_size);
+                    if (ret_r > 0) {
+                        ret_buf = process_protocol_buf(
+                            plugin, gd->cmd_sort->cmd[i].slave_id,
+                            response_size);
+                        if (ret_buf > 0) {
+                            rtt = neu_time_ms() - read_tms;
+                            break;
+                        } else if (ret_buf < 0) {
+                            if (ret_buf == -1) {
+                                modbus_value_handle(
+                                    plugin, gd->cmd_sort->cmd[i].slave_id, 0,
+                                    NULL,
+                                    NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE);
+                                plog_error(
+                                    plugin,
+                                    "modbus message error, skip, %hhu!%hu",
+                                    gd->cmd_sort->cmd[i].slave_id,
+                                    gd->cmd_sort->cmd[i].start_address);
+                            } else if (ret_buf == -2) {
+                                modbus_value_handle(
+                                    plugin, gd->cmd_sort->cmd[i].slave_id, 0,
+                                    NULL, NEU_ERR_PLUGIN_READ_FAILURE);
+                                plog_error(
+                                    plugin,
+                                    "modbus device response error, skip, "
+                                    "%hhu!%hu",
+                                    gd->cmd_sort->cmd[i].slave_id,
+                                    gd->cmd_sort->cmd[i].start_address);
+                            }
+                            rtt = neu_time_ms() - read_tms;
+                            break;
+                        }
+                    } else {
+                        modbus_value_handle(plugin,
+                                            gd->cmd_sort->cmd[i].slave_id, 0,
+                                            NULL, NEU_ERR_PLUGIN_DISCONNECTED);
+                        rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+                        neu_conn_disconnect(plugin->conn);
+                        break;
+                    }
+                }
+                if (ret_r > 0 && ret_buf == 0) {
+                    modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id,
+                                        0, NULL,
+                                        NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE);
+                    plog_warn(plugin,
+                              "no modbus response received, skip, %hhu!%hu",
+                              gd->cmd_sort->cmd[i].slave_id,
+                              gd->cmd_sort->cmd[i].start_address);
+                }
+            } else if (ret_buf < 0) {
+                if (ret_buf == -1) {
+                    modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id,
+                                        0, NULL,
+                                        NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE);
+                    plog_error(plugin, "modbus message error, skip, %hhu!%hu",
+                               gd->cmd_sort->cmd[i].slave_id,
+                               gd->cmd_sort->cmd[i].start_address);
+                } else if (ret_buf == -2) {
+                    modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id,
+                                        0, NULL, NEU_ERR_PLUGIN_READ_FAILURE);
+                    plog_error(plugin,
+                               "modbus device response error, skip, %hhu!%hu",
+                               gd->cmd_sort->cmd[i].slave_id,
+                               gd->cmd_sort->cmd[i].start_address);
+                }
+                rtt = neu_time_ms() - read_tms;
+            }
+        } else {
             for (uint16_t j = 0; j < plugin->max_retries; j++) {
-                ret = modbus_stack_read_retry(plugin, gd, i, j, &response_size);
-                if (ret > 0) {
-                    ret = process_protocol_buf(plugin, response_size);
-                    rtt = neu_time_ms() - read_tms;
-                    break;
+                ret_r =
+                    modbus_stack_read_retry(plugin, gd, i, j, &response_size);
+                if (ret_r > 0) {
+                    ret_buf = process_protocol_buf(
+                        plugin, gd->cmd_sort->cmd[i].slave_id, response_size);
+                    if (ret_buf > 0) {
+                        rtt = neu_time_ms() - read_tms;
+                        break;
+                    } else if (ret_buf < 0) {
+                        if (ret_buf == -1) {
+                            modbus_value_handle(
+                                plugin, gd->cmd_sort->cmd[i].slave_id, 0, NULL,
+                                NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE);
+                            plog_error(plugin,
+                                       "modbus message error, skip, %hhu!%hu",
+                                       gd->cmd_sort->cmd[i].slave_id,
+                                       gd->cmd_sort->cmd[i].start_address);
+                        } else if (ret_buf == -2) {
+                            modbus_value_handle(
+                                plugin, gd->cmd_sort->cmd[i].slave_id, 0, NULL,
+                                NEU_ERR_PLUGIN_READ_FAILURE);
+                            plog_error(plugin,
+                                       "modbus device response error, skip, "
+                                       "%hhu!%hu",
+                                       gd->cmd_sort->cmd[i].slave_id,
+                                       gd->cmd_sort->cmd[i].start_address);
+                        }
+                        rtt = neu_time_ms() - read_tms;
+                        break;
+                    } else if (ret_buf == 0) {
+                        modbus_value_handle(
+                            plugin, gd->cmd_sort->cmd[i].slave_id, 0, NULL,
+                            NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE);
+                        plog_warn(plugin,
+                                  "no modbus response received, skip, %hhu!%hu",
+                                  gd->cmd_sort->cmd[i].slave_id,
+                                  gd->cmd_sort->cmd[i].start_address);
+                        continue;
+                    }
                 }
             }
-            if (ret <= 0) {
+            if (ret_r <= 0) {
                 modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id, 0,
                                     NULL, NEU_ERR_PLUGIN_DISCONNECTED);
                 rtt = NEU_METRIC_LAST_RTT_MS_MAX;
                 neu_conn_disconnect(plugin->conn);
+                break;
             }
         }
         if (plugin->interval > 0) {
@@ -216,7 +330,7 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
     uint16_t start_address = gd->cmd_sort->cmd[plugin->cmd_idx].start_address;
     uint16_t n_register    = gd->cmd_sort->cmd[plugin->cmd_idx].n_register;
 
-    if (error != NEU_ERR_SUCCESS) {
+    if (error == NEU_ERR_PLUGIN_DISCONNECTED) {
         neu_dvalue_t dvalue = { 0 };
 
         dvalue.type      = NEU_TYPE_ERROR;
@@ -224,6 +338,17 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
         plugin->common.adapter_callbacks->driver.update(
             plugin->common.adapter, gd->group, NULL, dvalue);
         return 0;
+    } else if (error != NEU_ERR_SUCCESS) {
+        utarray_foreach(gd->cmd_sort->cmd[plugin->cmd_idx].tags,
+                        modbus_point_t **, p_tag)
+        {
+            neu_dvalue_t dvalue = { 0 };
+            dvalue.type         = NEU_TYPE_ERROR;
+            dvalue.value.i32    = error;
+            plugin->common.adapter_callbacks->driver.update(
+                plugin->common.adapter, gd->group, (*p_tag)->name, dvalue);
+        }
+        return 0;
     }
 
     utarray_foreach(gd->cmd_sort->cmd[plugin->cmd_idx].tags, modbus_point_t **,
@@ -325,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);
@@ -378,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, response_size);
+        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)
@@ -409,29 +605,104 @@ static void plugin_group_free(neu_plugin_group_t *pgp)
     free(gd);
 }
 
-static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size)
+static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
+                                uint16_t response_size)
 {
     uint8_t *                 recv_buf = calloc(response_size, 1);
     neu_protocol_unpack_buf_t pbuf     = { 0 };
     ssize_t                   ret      = 0;
-
-    if (plugin->is_server && plugin->protocol == MODBUS_PROTOCOL_TCP) {
-        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 == response_size) {
-        if (response_size < 512) {
-            plog_recv_protocol(plugin, recv_buf, ret);
+    if (plugin->protocol == MODBUS_PROTOCOL_TCP) {
+        if (plugin->is_server) {
+            ret = neu_conn_tcp_server_recv(plugin->conn, plugin->client_fd,
+                                           recv_buf,
+                                           sizeof(struct modbus_header));
+        } else {
+            ret = neu_conn_recv(plugin->conn, recv_buf,
+                                sizeof(struct modbus_header));
         }
-        neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
-        ret = modbus_stack_recv(plugin->stack, &pbuf);
-    } else {
-        ret = -1;
-    }
 
+        if (ret == 0 || ret == -1) {
+            free(recv_buf);
+            return 0;
+        }
+
+        if (ret == sizeof(struct modbus_header)) {
+            struct modbus_header *header = (struct modbus_header *) recv_buf;
+            int                   ret1   = 0;
+
+            if (htons(header->len) >
+                response_size - sizeof(struct modbus_header)) {
+                free(recv_buf);
+                return -1;
+            }
+
+            plog_recv_protocol(plugin, recv_buf, ret);
+            if (plugin->is_server) {
+                ret1 = neu_conn_tcp_server_recv(
+                    plugin->conn, plugin->client_fd,
+                    recv_buf + sizeof(struct modbus_header),
+                    htons(header->len));
+            } else {
+                ret1 = neu_conn_recv(plugin->conn,
+                                     recv_buf + sizeof(struct modbus_header),
+                                     htons(header->len));
+            }
+            if (ret1 != htons(header->len)) {
+                free(recv_buf);
+                return -1;
+            }
+            ret += ret1;
+        } else {
+            free(recv_buf);
+            return -1;
+        }
+
+        if (ret > 0) {
+            if (ret < 512) {
+                plog_recv_protocol(plugin, recv_buf, ret);
+            }
+            neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
+            int ret_s = modbus_stack_recv(plugin->stack, slave_id, &pbuf);
+            if (ret_s == MODBUS_DEVICE_ERR) {
+                ret = ret_s;
+            } else {
+                if (ret != response_size) {
+                    ret = -1;
+                } else {
+                    ret = ret_s;
+                }
+            }
+        }
+    } else if (plugin->protocol == MODBUS_PROTOCOL_RTU) {
+        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);
+            return 0;
+        }
+
+        if (ret > 0) {
+            if (response_size < 512) {
+                plog_recv_protocol(plugin, recv_buf, ret);
+            }
+            neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
+            int ret_s = modbus_stack_recv(plugin->stack, slave_id, &pbuf);
+            if (ret_s == MODBUS_DEVICE_ERR) {
+                ret = ret_s;
+            } else {
+                if (ret != response_size) {
+                    ret = -1;
+                } else {
+                    ret = ret_s;
+                }
+            }
+        }
+    }
     free(recv_buf);
     return ret;
 }

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 = {
@@ -66,11 +70,11 @@ const neu_plugin_module_t neu_plugin_module = {
     .module_name = "Modbus RTU",
     .module_descr =
         "This plugin is used to connect devices using modbus rtu protocol. "
-        "Support serial,TCP and UDP communication methods. Selecting "
+        "Support serial and TCP communication methods. Selecting "
         "client/server is also supported.",
     .module_descr_zh =
-        "该插件用于连接使用 modbus rtu 协议的设备。支持串口, TCP 和 UDP "
-        "三种通信方式。也支持选择客户端/服务端。",
+        "该插件用于连接使用 modbus rtu 协议的设备。支持串口, TCP "
+        "两种通信方式。也支持选择客户端/服务端。",
     .intf_funs = &plugin_intf_funs,
     .kind      = NEU_PLUGIN_KIND_SYSTEM,
     .type      = NEU_NA_TYPE_DRIVER,
@@ -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

@@ -62,7 +62,8 @@ void modbus_stack_destroy(modbus_stack_t *stack)
     free(stack);
 }
 
-int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
+int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
+                      neu_protocol_unpack_buf_t *buf)
 {
     struct modbus_header header = { 0 };
     struct modbus_code   code   = { 0 };
@@ -74,13 +75,17 @@ int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
             plog_warn((neu_plugin_t *) stack->ctx, "try modbus rtu driver");
         }
         if (ret <= 0) {
-            return ret;
+            return -1;
         }
     }
 
     ret = modbus_code_unwrap(buf, &code);
     if (ret <= 0) {
-        return ret;
+        return -1;
+    }
+
+    if (code.slave_id != slave_id) {
+        return -1;
     }
 
     switch (code.function) {
@@ -92,7 +97,7 @@ int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
         uint8_t *          bytes = NULL;
         ret                      = modbus_data_unwrap(buf, &data);
         if (ret <= 0) {
-            return ret;
+            return -1;
         }
 
         if (stack->protocol == MODBUS_PROTOCOL_TCP && data.n_byte == 0xff) {
@@ -116,18 +121,38 @@ int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
     case MODBUS_WRITE_M_COIL: {
         struct modbus_address address = { 0 };
         ret                           = modbus_address_unwrap(buf, &address);
+        if (ret <= 0) {
+            return -1;
+        }
         break;
     }
+    case MODBUS_READ_COIL_ERR:
+        return MODBUS_DEVICE_ERR;
+    case MODBUS_READ_INPUT_ERR:
+        return MODBUS_DEVICE_ERR;
+    case MODBUS_READ_HOLD_REG_ERR:
+        return MODBUS_DEVICE_ERR;
+    case MODBUS_READ_INPUT_REG_ERR:
+        return MODBUS_DEVICE_ERR;
+    case MODBUS_WRITE_S_COIL_ERR:
+        return MODBUS_DEVICE_ERR;
+    case MODBUS_WRITE_S_HOLD_REG_ERR:
+        return MODBUS_DEVICE_ERR;
+    case MODBUS_WRITE_M_HOLD_REG_ERR:
+        return MODBUS_DEVICE_ERR;
+    case MODBUS_WRITE_M_COIL_ERR:
+        return MODBUS_DEVICE_ERR;
     case MODBUS_WRITE_S_HOLD_REG:
-    default:
         break;
+    default:
+        return -1;
     }
 
     if (stack->protocol == MODBUS_PROTOCOL_RTU) {
         struct modbus_crc crc = { 0 };
         ret                   = modbus_crc_unwrap(buf, &crc);
         if (ret <= 0) {
-            return ret;
+            return -1;
         }
     }
 
@@ -138,10 +163,11 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
                       enum modbus_area area, uint16_t start_address,
                       uint16_t n_reg, uint16_t *response_size)
 {
-    static __thread uint8_t                 buf[16]  = { 0 };
-    static __thread neu_protocol_pack_buf_t pbuf     = { 0 };
-    int                                     ret      = 0;
-    modbus_action_e                         m_action = MODBUS_ACTION_DEFAULT;
+    static __thread uint8_t                 buf[16] = { 0 };
+    static __thread neu_protocol_pack_buf_t pbuf    = { 0 };
+    int                                     ret     = 0;
+    *response_size                                  = 0;
+    modbus_action_e m_action                        = MODBUS_ACTION_DEFAULT;
 
     neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
 
@@ -197,7 +223,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;
@@ -211,13 +237,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);
@@ -247,13 +281,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

@@ -43,7 +43,8 @@ modbus_stack_t *modbus_stack_create(void *ctx, modbus_protocol_e protocol,
                                     modbus_stack_write_resp write_resp);
 void            modbus_stack_destroy(modbus_stack_t *stack);
 
-int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf);
+int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
+                      neu_protocol_unpack_buf_t *buf);
 
 int  modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
                        enum modbus_area area, uint16_t start_address,
@@ -51,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)
@@ -244,7 +274,7 @@ int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
     return 0;
 }
 
-void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
+void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state)
 {
     neu_reqresp_head_t  header   = { .type = NEU_REQ_NODE_INIT };
     neu_req_node_init_t init     = { 0 };
@@ -254,8 +284,8 @@ void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
     strcpy(header.receiver, "manager");
 
     strcpy(init.node, adapter->name);
-    init.auto_start = auto_start;
-    init_msg        = neu_msg_gen(&header, &init);
+    init.state = state;
+    init_msg   = neu_msg_gen(&header, &init);
 
     nng_sendmsg(adapter->sock, init_msg, 0);
 }
@@ -321,6 +351,22 @@ static int adapter_update_metric(neu_adapter_t *adapter,
     return 0;
 }
 
+static void adapter_reset_metrics(neu_adapter_t *adapter)
+{
+    neu_metric_entry_t *entry = NULL;
+    if (NULL == adapter->metrics) {
+        return;
+    }
+
+    HASH_LOOP(hh, adapter->metrics->entries, entry) { entry->value = 0; }
+
+    neu_group_metrics_t *g = NULL;
+    HASH_LOOP(hh, adapter->metrics->group_metrics, g)
+    {
+        HASH_LOOP(hh, g->entries, entry) { entry->value = 0; }
+    }
+}
+
 static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
                            void *data)
 {
@@ -766,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 {
@@ -780,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]);
         }
@@ -1056,6 +1124,7 @@ int neu_adapter_stop(neu_adapter_t *adapter)
     if (error == NEU_ERR_SUCCESS) {
         adapter->state = NEU_NODE_RUNNING_STATE_STOPPED;
         adapter_storage_state(adapter->name, adapter->state);
+        adapter_reset_metrics(adapter);
     }
 
     return error;

src/adapter/adapter_internal.h

@@ -63,8 +63,11 @@ typedef struct adapter_msg_handler {
     adapter_handler    handler;
 } adapter_msg_handler_t;
 
-neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
-void           neu_adapter_init(neu_adapter_t *adapter, bool auto_start);
+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/cache.c

@@ -123,9 +123,10 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
     nng_mtx_unlock(cache->mtx);
 }
 
-void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
-                             const char *tag, int64_t timestamp,
-                             neu_dvalue_t value)
+void neu_driver_cache_update_change(neu_driver_cache_t *cache,
+                                    const char *group, const char *tag,
+                                    int64_t timestamp, neu_dvalue_t value,
+                                    bool change)
 {
     struct elem *elem = NULL;
     tkey_t       key  = to_key(group, tag);
@@ -185,6 +186,9 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
             }
         }
 
+        if (change) {
+            elem->changed = true;
+        }
         elem->value.type  = value.type;
         elem->value.value = value.value;
     }
@@ -192,6 +196,13 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
     nng_mtx_unlock(cache->mtx);
 }
 
+void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
+                             const char *tag, int64_t timestamp,
+                             neu_dvalue_t value)
+{
+    neu_driver_cache_update_change(cache, group, tag, timestamp, value, false);
+}
+
 int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
                          const char *tag, neu_driver_cache_value_t *value)
 {

src/adapter/driver/cache.h

@@ -34,6 +34,10 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
 void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
                              const char *tag, int64_t timestamp,
                              neu_dvalue_t value);
+void neu_driver_cache_update_change(neu_driver_cache_t *cache,
+                                    const char *group, const char *tag,
+                                    int64_t timestamp, neu_dvalue_t value,
+                                    bool change);
 
 void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
                           const char *tag);

src/adapter/driver/driver.c

@@ -103,10 +103,59 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 static void update_im(neu_adapter_t *adapter, const char *group,
                       const char *tag, neu_dvalue_t value)
 {
-    (void) adapter;
-    (void) group;
-    (void) tag;
-    (void) value;
+    neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
+    neu_reqresp_head_t    header = { 0 };
+
+    if (tag == NULL || value.type == NEU_TYPE_ERROR) {
+        nlog_warn("update_im tag is null or value is error %d",
+                  value.value.i32);
+        return;
+    }
+
+    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,
+                                          NEU_METRIC_TAG_READS_TOTAL, 1, NULL);
+    neu_datatag_t *first = utarray_front(tags);
+
+    if (first == NULL ||
+        !neu_tag_attribute_test(first, NEU_ATTRIBUTE_SUBSCRIBE)) {
+        utarray_free(tags);
+        nlog_info("update immediately, driver: %s, "
+                  "group: %s, tag: %s, type: %s, "
+                  "timestamp: %" PRId64,
+                  driver->adapter.name, group, tag, neu_type_string(value.type),
+                  global_timestamp);
+        return;
+    }
+
+    nlog_info("update and report immediately, driver: %s, "
+              "group: %s, tag: %s, type: %s, "
+              "timestamp: %" PRId64,
+              driver->adapter.name, group, tag, neu_type_string(value.type),
+              global_timestamp);
+
+    header.type                    = NEU_REQRESP_TRANS_DATA;
+    neu_reqresp_trans_data_t *data = calloc(
+        1, sizeof(neu_reqresp_trans_data_t) + sizeof(neu_resp_tag_value_t));
+
+    strcpy(data->driver, driver->adapter.name);
+    strcpy(data->group, group);
+
+    data->n_tag = read_report_group(global_timestamp, 0, driver->cache, group,
+                                    tags, (neu_resp_tag_value_t *) &data[1]);
+
+    if (data->n_tag > 0) {
+        driver->adapter.cb_funs.response(&driver->adapter, &header, data);
+    }
+
+    utarray_free(tags);
+    free(data);
 }
 
 static void update(neu_adapter_t *adapter, const char *group, const char *tag,
@@ -197,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);
@@ -334,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);
@@ -342,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:
@@ -358,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);
@@ -366,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:
@@ -408,7 +458,7 @@ static void cal_decimal(neu_type_e tag_type, neu_type_e value_type,
     case NEU_TYPE_WORD:
     case NEU_TYPE_LWORD:
     case NEU_TYPE_DWORD:
-        if (value_type == NEU_TYPE_INT64 || NEU_TYPE_LWORD) {
+        if (value_type == NEU_TYPE_INT64 || value_type == NEU_TYPE_LWORD) {
             value->u64 = value->u64 / decimal;
         } else if (value_type == NEU_TYPE_DOUBLE) {
             value->u64 = (uint64_t) round(value->d64 / decimal);
@@ -477,7 +527,7 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
 
             if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
                 if (cmd->tags[i].value.type == NEU_TYPE_INT64) {
-                    tv.value.d64 = (double) tv.value.u64;
+                    tv.value.d64 = (double) tv.value.i64;
                 }
             }
             if (tag->decimal != 0) {
@@ -488,10 +538,9 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
 
             tv.value = cmd->tags[i].value.value;
             utarray_push_back(tags, &tv);
-        } else {
-            if (tag != NULL) {
-                neu_tag_free(tag);
-            }
+        }
+        if (tag != NULL) {
+            neu_tag_free(tag);
         }
     }
 
@@ -554,7 +603,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
         }
         if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
             if (cmd->value.type == NEU_TYPE_INT64) {
-                cmd->value.value.d64 = (double) cmd->value.value.u64;
+                cmd->value.value.d64 = (double) cmd->value.value.i64;
             }
         }
 
@@ -689,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);
@@ -725,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;
     }
@@ -779,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;
@@ -807,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;
@@ -816,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);
@@ -848,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));
@@ -973,6 +1068,25 @@ UT_array *neu_adapter_driver_get_read_tag(neu_adapter_driver_t *driver,
     return tags;
 }
 
+UT_array *neu_adapter_driver_get_ptag(neu_adapter_driver_t *driver,
+                                      const char *group, const char *tag)
+{
+    group_t * find = NULL;
+    UT_array *tags = NULL;
+
+    HASH_FIND_STR(driver->groups, group, find);
+    if (find != NULL) {
+        neu_datatag_t *t = neu_group_find_tag(find->group, tag);
+        if (t != NULL) {
+            utarray_new(tags, neu_tag_get_icd());
+            utarray_push_back(tags, t);
+            neu_tag_free(t);
+        }
+    }
+
+    return tags;
+}
+
 static int report_callback(void *usr_data)
 {
     group_t *                group  = (group_t *) usr_data;
@@ -1123,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(
@@ -1230,7 +1344,7 @@ static int read_report_group(int64_t timestamp, int64_t timeout,
         }
 
         if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
-            (timestamp - value.timestamp) > timeout) {
+            (timestamp - value.timestamp) > timeout && timeout > 0) {
             datas[index].value.type      = NEU_TYPE_ERROR;
             datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED;
         } else {

src/adapter/driver/driver_internal.h

@@ -51,19 +51,31 @@ 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_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,
-                                const char *tag);
-int  neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
-                                   const char *group, neu_datatag_t *tag);
-int  neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group,
-                                UT_array **tags);
-int  neu_adapter_driver_query_tag(neu_adapter_driver_t *driver,
-                                  const char *group, const char *name,
-                                  UT_array **tags);
-void neu_adapter_driver_get_value_tag(neu_adapter_driver_t *driver,
-                                      const char *group, UT_array **tags);
+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,
+                               const char *tag);
+int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
+                                  const char *group, neu_datatag_t *tag);
+int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group,
+                               UT_array **tags);
+UT_array *neu_adapter_driver_get_ptag(neu_adapter_driver_t *driver,
+                                      const char *group, const char *tag);
+int       neu_adapter_driver_query_tag(neu_adapter_driver_t *driver,
+                                       const char *group, const char *name,
+                                       UT_array **tags);
+void      neu_adapter_driver_get_value_tag(neu_adapter_driver_t *driver,
+                                           const char *group, UT_array **tags);
 UT_array *neu_adapter_driver_get_read_tag(neu_adapter_driver_t *driver,
                                           const char *          group);
 #endif

src/adapter/storage.c

@@ -125,7 +125,7 @@ int adapter_load_setting(const char *node, char **setting)
 {
     int rv = neu_persister_load_node_setting(node, (const char **) setting);
     if (0 != rv) {
-        nlog_error("load %s setting fail", node);
+        nlog_warn("load %s setting fail", node);
         return -1;
     }
 
@@ -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,6 +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"
 "    --log                log to the stdout\n"
 "    --log_level <LEVEL>  default log level(DEBUG,NOTICE)\n"
 "    --reset-password     reset dashboard to use default password\n"
@@ -70,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";
@@ -132,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;
@@ -149,8 +338,10 @@ 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' },
         { NULL, 0, NULL, 0 },
     };
 
@@ -159,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) {
@@ -175,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':
@@ -195,9 +406,17 @@ 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 '?':
@@ -273,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
@@ -46,6 +56,9 @@ typedef struct {
     char * log_init_file;
     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

@@ -274,10 +274,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             break;
         }
 
-        if (init->auto_start) {
+        if (init->state == NEU_NODE_RUNNING_STATE_RUNNING ||
+            init->state == NEU_NODE_RUNNING_STATE_STOPPED) {
             neu_adapter_t *adapter =
                 neu_node_manager_find(manager->node_manager, init->node);
             neu_adapter_start(adapter);
+            if (init->state == NEU_NODE_RUNNING_STATE_STOPPED) {
+                neu_adapter_stop(adapter);
+            }
         }
 
         nlog_notice("bind node %s to pipe(%d)", init->node, pipe.id);
@@ -560,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) {
@@ -1092,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);
@@ -1115,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"
@@ -45,7 +46,8 @@ 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, bool start)
+                         const char *             plugin_name,
+                         neu_node_running_state_e state, bool load)
 {
     neu_adapter_t *       adapter      = NULL;
     neu_plugin_instance_t instance     = { 0 };
@@ -77,9 +79,12 @@ 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, start);
+    neu_adapter_init(adapter, state);
 
     return NEU_ERR_SUCCESS;
 }
@@ -516,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;
     }
@@ -616,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

@@ -52,7 +52,8 @@ int       neu_manager_del_plugin(neu_manager_t *manager, const char *plugin);
 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, bool start);
+                               const char *             plugin_name,
+                               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

@@ -362,9 +362,9 @@ 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 == NEU_NODE_RUNNING_STATE_RUNNING);
+        rv                    = neu_manager_add_node(manager, node_info->name,
+                                  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;
@@ -135,3 +178,40 @@ int neuron_already_running()
 
     return 0;
 }
+
+int neuron_stop()
+{
+    int ret = -1;
+
+    FILE *fp = fopen(NEURON_DAEMON_LOCK_FNAME, "r");
+    if (NULL == fp) {
+        nlog_error("cannot open %s reason: %s\n", NEURON_DAEMON_LOCK_FNAME,
+                   strerror(errno));
+        return ret;
+    }
+
+    long pid = -1;
+    if (1 != fscanf(fp, "%ld", &pid)) {
+        nlog_error("cannot scan pid from: %s\n", NEURON_DAEMON_LOCK_FNAME);
+        goto end;
+    }
+
+    pid_t gid = getpgid((pid_t) pid);
+    if (-1 == gid) {
+        nlog_error("cannot get gpid reason: %s\n", strerror(errno));
+        goto end;
+    }
+
+    if (0 == kill((pid_t)(-gid), SIGINT)) {
+        if (neuron_wait_exit(STOP_TIMEOUT) == 0) {
+            ret = 0;
+        }
+    } else {
+        nlog_error("cannot kill gpid:%ld reason: %s\n", (long) gid,
+                   strerror(errno));
+    }
+
+end:
+    fclose(fp);
+    return ret;
+}

src/daemon.h

@@ -34,6 +34,8 @@ void daemonize();
  */
 int neuron_already_running();
 
+int neuron_stop();
+
 #ifdef __cplusplus
 }
 #endif

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`
@@ -110,9 +112,25 @@ int main(int argc, char *argv[])
     zlog_level_switch(neuron, default_log_level);
 
     if (neuron_already_running()) {
-        nlog_fatal("neuron process already running, exit.");
-        rv = -1;
+        if (args.stop) {
+            rv = neuron_stop();
+            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.");
+            rv = -1;
+        }
         goto main_end;
+    } else {
+        if (args.stop) {
+            rv = 0;
+            printf("neuron no running.\n");
+            goto main_end;
+        }
     }
 
     for (size_t i = 0; i < args.restart; ++i) {

src/persist/persist.c

@@ -54,8 +54,8 @@ static const char *plugin_file       = "persistence/plugins.json";
 static const char *db_file           = "persistence/sqlite.db";
 static sqlite3 *   global_db         = NULL;
 pthread_rwlock_t   global_rwlock     = PTHREAD_RWLOCK_INITIALIZER;
-static size_t      global_node_count = 0;
-static size_t      global_tag_count  = 0;
+static int         global_node_count = 0;
+static int         global_tag_count  = 0;
 
 static inline bool ends_with(const char *str, const char *suffix)
 {
@@ -486,7 +486,9 @@ static inline int node_count_load()
 
     if (SQLITE_OK == sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL) &&
         SQLITE_ROW == sqlite3_step(stmt)) {
+        pthread_rwlock_wrlock(&global_rwlock);
         global_node_count = sqlite3_column_int(stmt, 0);
+        pthread_rwlock_unlock(&global_rwlock);
     } else {
         rv = -1;
     }
@@ -503,7 +505,9 @@ static inline int tag_count_load()
 
     if (SQLITE_OK == sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL) &&
         SQLITE_ROW == sqlite3_step(stmt)) {
+        pthread_rwlock_wrlock(&global_rwlock);
         global_tag_count = sqlite3_column_int(stmt, 0);
+        pthread_rwlock_unlock(&global_rwlock);
     } else {
         rv = -1;
     }
@@ -595,13 +599,22 @@ void neu_persister_destroy()
 
 int neu_persister_store_node(neu_persist_node_info_t *info)
 {
-    int rv =
-        execute_sql(global_db,
-                    "INSERT INTO nodes (name, type, state, plugin_name) "
-                    "VALUES (%Q, %i, %i, %Q)",
-                    info->name, info->type, info->state, info->plugin_name);
-    if (0 == rv) {
-        node_count_add(1);
+    int rv = 0;
+    if (strcmp(info->plugin_name, "Monitor") == 0) {
+        rv = execute_sql(
+            global_db,
+            "INSERT OR IGNORE INTO nodes (name, type, state, plugin_name) "
+            "VALUES (%Q, %i, %i, %Q)",
+            info->name, info->type, info->state, info->plugin_name);
+    } else {
+        rv =
+            execute_sql(global_db,
+                        "INSERT INTO nodes (name, type, state, plugin_name) "
+                        "VALUES (%Q, %i, %i, %Q)",
+                        info->name, info->type, info->state, info->plugin_name);
+        if (0 == rv) {
+            node_count_add(1);
+        }
     }
     return rv;
 }
@@ -668,6 +681,7 @@ int neu_persister_delete_node(const char *node_name)
         execute_sql(global_db, "DELETE FROM nodes WHERE name=%Q;", node_name);
     if (0 == rv) {
         node_count_add(-1);
+        tag_count_load();
     }
     return rv;
 }
@@ -1219,9 +1233,13 @@ error:
 int neu_persister_delete_group(const char *driver_name, const char *group_name)
 {
     // rely on foreign key constraints to delete tags and subscriptions
-    return execute_sql(global_db,
-                       "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
-                       driver_name, group_name);
+    int rv = execute_sql(global_db,
+                         "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
+                         driver_name, group_name);
+    if (0 == rv) {
+        tag_count_load();
+    }
+    return rv;
 }
 
 int neu_persister_store_node_setting(const char *node_name, const char *setting)

tests/CMakeLists.txt

@@ -42,9 +42,17 @@ target_include_directories(tag_sort_test PRIVATE
 target_link_libraries(tag_sort_test neuron-base gtest_main gtest)
 #target_link_directories(modbus_point_test PRIVATE /usr/local/lib)
 
+add_executable(modbus_test modbus_test.cc
+				${CMAKE_SOURCE_DIR}/plugins/modbus/modbus.c
+				${CMAKE_SOURCE_DIR}/plugins/modbus/modbus_point.c)
+target_include_directories(modbus_test PRIVATE
+				${CMAKE_SOURCE_DIR}/plugins/modbus)
+target_link_libraries(modbus_test neuron-base gtest_main gtest pthread zlog)
+
 include(GoogleTest)
 gtest_discover_tests(json_test)
 gtest_discover_tests(http_test)
 gtest_discover_tests(jwt_test)
 gtest_discover_tests(base64_test)
-gtest_discover_tests(tag_sort_test)
+gtest_discover_tests(tag_sort_test)
+gtest_discover_tests(modbus_test)

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:

tests/modbus_test.cc

@@ -0,0 +1,163 @@
+#include <cstdint>
+#include <gtest/gtest.h>
+#include <neuron.h>
+extern "C" {
+#include "modbus.h"
+#include "modbus_point.h"
+}
+
+zlog_category_t *neuron           = NULL;
+int64_t          global_timestamp = 0;
+
+TEST(test_modbus_header_wrap, should_return_right_header_value)
+{
+    uint8_t                 buf[16] = { 0 };
+    neu_protocol_pack_buf_t pbuf    = { 0 };
+    uint16_t                seq     = 0x0004;
+
+    neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_header_wrap(&pbuf, seq);
+
+    struct modbus_header *header =
+        (struct modbus_header *) neu_protocol_unpack_buf(
+            &pbuf, sizeof(struct modbus_header));
+
+    EXPECT_EQ(0x0004, ntohs(header->seq));
+    EXPECT_EQ(0x0, header->protocol);
+    EXPECT_EQ(0x0, ntohs(header->len));
+}
+
+TEST(test_modbus_header_unwrap, should_return_right_header_value)
+{
+    uint8_t                   buf[16] = { 0x00, 0x01, 0x00, 0x00, 0x00, 0x00 };
+    neu_protocol_unpack_buf_t pbuf    = { 0 };
+    struct modbus_header      header  = { 0 };
+
+    neu_protocol_unpack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_header_unwrap(&pbuf, &header);
+
+    EXPECT_EQ(0x1, header.seq);
+    EXPECT_EQ(0x0, header.protocol);
+    EXPECT_EQ(0x0, header.len);
+}
+
+TEST(test_modbus_code_wrap, should_return_right_code_value)
+{
+    uint8_t                 buf[16]  = { 0 };
+    neu_protocol_pack_buf_t pbuf     = { 0 };
+    uint8_t                 slave_id = 1;
+    uint8_t                 function = MODBUS_READ_HOLD_REG;
+    neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_code_wrap(&pbuf, slave_id, function);
+
+    struct modbus_code *code = (struct modbus_code *) neu_protocol_unpack_buf(
+        &pbuf, sizeof(struct modbus_code));
+
+    EXPECT_EQ(0x1, code->slave_id);
+    EXPECT_EQ(0x3, code->function);
+}
+
+TEST(test_modbus_code_unwrap, should_return_right_code_value)
+{
+    uint8_t                   buf[16] = { 0x01, 0x03 };
+    neu_protocol_unpack_buf_t pbuf    = { 0 };
+    struct modbus_code        code    = { 0 };
+
+    neu_protocol_unpack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_code_unwrap(&pbuf, &code);
+
+    EXPECT_EQ(0x1, code.slave_id);
+    EXPECT_EQ(0x3, code.function);
+}
+
+TEST(test_modbus_address_wrap, should_return_right_address_value)
+{
+    uint8_t                 buf[16]    = { 0 };
+    neu_protocol_pack_buf_t pbuf       = { 0 };
+    uint16_t                start      = 0x1;
+    uint16_t                n_register = 0x2;
+    enum modbus_action      action     = MODBUS_ACTION_DEFAULT;
+
+    neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_address_wrap(&pbuf, start, n_register, action);
+
+    struct modbus_address *address =
+        (struct modbus_address *) neu_protocol_unpack_buf(
+            &pbuf, sizeof(struct modbus_address));
+
+    EXPECT_EQ(0x1, ntohs(address->start_address));
+    EXPECT_EQ(0x2, ntohs(address->n_reg));
+}
+
+TEST(test_modbus_address_unwrap, should_return_right_address_value)
+{
+    uint8_t                   buf[16]     = { 0x00, 0x01, 0x00, 0x02 };
+    neu_protocol_unpack_buf_t pbuf        = { 0 };
+    struct modbus_address     out_address = { 0 };
+
+    neu_protocol_unpack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_address_unwrap(&pbuf, &out_address);
+
+    struct modbus_address *address =
+        (struct modbus_address *) neu_protocol_pack_buf(
+            &pbuf, sizeof(struct modbus_address));
+
+    EXPECT_EQ(0x1, ntohs(address->start_address));
+    EXPECT_EQ(0x2, ntohs(address->n_reg));
+}
+
+TEST(test_modbus_data_wrap, should_return_right_data_value)
+{
+    uint8_t                 buf[16]  = { 0 };
+    neu_protocol_pack_buf_t pbuf     = { 0 };
+    uint8_t                 n_byte   = 4;
+    uint8_t                 byte[16] = { "\x11\x22\x33\x44" };
+    uint8_t *               bytes    = byte;
+    enum modbus_action      action   = MODBUS_ACTION_DEFAULT;
+
+    neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_data_wrap(&pbuf, n_byte, bytes, action);
+
+    struct modbus_data *data = (struct modbus_data *) neu_protocol_unpack_buf(
+        &pbuf, sizeof(struct modbus_data));
+
+    EXPECT_EQ(n_byte, data->n_byte);
+    EXPECT_EQ(0x11, data->byte[0]);
+    EXPECT_EQ(0x22, data->byte[1]);
+    EXPECT_EQ(0x33, data->byte[2]);
+    EXPECT_EQ(0x44, data->byte[3]);
+}
+
+TEST(test_modbus_data_unwrap, should_return_right_data_value)
+{
+    uint8_t                   buf[16] = { "\x04\x11\x22\x33\x44" };
+    neu_protocol_unpack_buf_t pbuf    = { 0 };
+    struct modbus_data        data    = { 0 };
+    uint8_t *                 bytes   = NULL;
+
+    neu_protocol_unpack_buf_init(&pbuf, buf, sizeof(buf));
+
+    modbus_data_unwrap(&pbuf, &data);
+    bytes = neu_protocol_unpack_buf(&pbuf, 4);
+
+    EXPECT_EQ(4, data.n_byte);
+    EXPECT_EQ(0x11, *bytes);
+    EXPECT_EQ(0x22, *(bytes + 1));
+    EXPECT_EQ(0x33, *(bytes + 2));
+    EXPECT_EQ(0x44, *(bytes + 3));
+}
+
+int main(int argc, char **argv)
+{
+    zlog_init("./config/dev.conf");
+    neuron = zlog_get_category("neuron");
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

version

@@ -1 +1 @@
-2.5.2
+2.5.5