Merge pull request #2520 from fengzeroz/v2.12

version: 2.12.1

.github/workflows/check.yml

@@ -27,7 +27,7 @@ jobs:
 
   cppcheck:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.12
 
     steps:
       - uses: actions/checkout@v4
@@ -42,7 +42,7 @@ jobs:
 
   build:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.12
 
     steps:
       - uses: actions/checkout@v4
@@ -65,7 +65,7 @@ jobs:
 
   ut:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.12
     needs: [build]
 
     steps:
@@ -103,7 +103,7 @@ jobs:
 
   pft:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.12
     needs: [build]
     strategy:
       matrix:

.github/workflows/ci.yml

@@ -8,9 +8,9 @@ on:
       - published
 
 env:
-  VERSION: 2.12.0
+  VERSION: 2.12.1
   UI_VERSION: 2.6.1
-  BRANCH: main
+  BRANCH: v2.12
 
 jobs:
   cross-build:
@@ -19,7 +19,7 @@ jobs:
       matrix:
         arch:
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.12
 
     steps:
       - uses: actions/checkout@v4
@@ -62,7 +62,7 @@ jobs:
       matrix:
         arch:
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.12
 
     steps:
       - uses: actions/checkout@v4

cmake/aarch64-linux-gnu.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX aarch64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR aarch64)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.12/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/arm-linux-gnueabi.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabi)
 set(CMAKE_SYSTEM_PROCESSOR armv5tejl)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.12/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/arm-linux-gnueabihf.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabihf)
 set(CMAKE_SYSTEM_PROCESSOR armv7l)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.12/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/x86_64-linux-gnu.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX x86_64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR x86_64)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.12/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 12
-#define NEU_VERSION_FIX 0
+#define NEU_VERSION_FIX 1
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)

include/neuron/msg.h

@@ -68,6 +68,8 @@ typedef enum neu_reqresp_type {
     NEU_RESP_GET_DATALAYERS_GROUPS,
     NEU_REQ_GET_DATALAYERS_TAGS,
     NEU_RESP_GET_DATALAYERS_TAGS,
+    NEU_REQ_GET_DATALAYERS_TAG,
+    NEU_RESP_GET_DATALAYERS_TAG,
 
     NEU_REQ_NODE_INIT,
     NEU_REQ_NODE_UNINIT,
@@ -199,6 +201,8 @@ static const char *neu_reqresp_type_string_t[] = {
     [NEU_RESP_GET_DATALAYERS_GROUPS] = "NEU_RESP_GET_DATALAYERS_GROUPS",
     [NEU_REQ_GET_DATALAYERS_TAGS]    = "NEU_REQ_GET_DATALAYERS_TAGS",
     [NEU_RESP_GET_DATALAYERS_TAGS]   = "NEU_RESP_GET_DATALAYERS_TAGS",
+    [NEU_REQ_GET_DATALAYERS_TAG]     = "NEU_REQ_GET_DATALAYERS_TAG",
+    [NEU_RESP_GET_DATALAYERS_TAG]    = "NEU_RESP_GET_DATALAYERS_TAG",
 
     [NEU_REQ_NODE_INIT]         = "NEU_REQ_NODE_INIT",
     [NEU_REQ_NODE_UNINIT]       = "NEU_REQ_NODE_UNINIT",
@@ -697,6 +701,9 @@ typedef struct neu_req_get_subscribe_group {
 
 typedef struct {
     char app[NEU_NODE_NAME_LEN];
+    char driver[NEU_NODE_NAME_LEN];
+    char group[NEU_GROUP_NAME_LEN];
+    char tag[NEU_TAG_NAME_LEN];
 } neu_req_get_sub_driver_tags_t;
 
 typedef struct neu_resp_subscribe_info {
@@ -881,10 +888,11 @@ typedef struct neu_resp_tag_value {
 typedef neu_resp_tag_value_t neu_tag_value_t;
 
 typedef struct neu_resp_tag_value_meta {
-    char           tag[NEU_TAG_NAME_LEN];
-    neu_dvalue_t   value;
-    neu_tag_meta_t metas[NEU_TAG_META_SIZE];
-    neu_datatag_t  datatag;
+    char            tag[NEU_TAG_NAME_LEN];
+    neu_dvalue_t    value;
+    neu_tag_meta_t *metas;
+    int             n_meta;
+    neu_datatag_t   datatag;
 } neu_resp_tag_value_meta_t;
 
 static inline UT_icd *neu_resp_tag_value_meta_icd()
@@ -955,10 +963,11 @@ typedef struct {
 } neu_resp_read_group_t;
 
 typedef struct neu_resp_tag_value_meta_paginate {
-    char           tag[NEU_TAG_NAME_LEN];
-    neu_dvalue_t   value;
-    neu_tag_meta_t metas[NEU_TAG_META_SIZE];
-    neu_datatag_t  datatag;
+    char            tag[NEU_TAG_NAME_LEN];
+    neu_dvalue_t    value;
+    neu_tag_meta_t *metas;
+    int             n_meta;
+    neu_datatag_t   datatag;
 } neu_resp_tag_value_meta_paginate_t;
 
 static inline UT_icd *neu_resp_tag_value_meta_paginate_icd()
@@ -989,6 +998,13 @@ static inline void neu_resp_read_free(neu_resp_read_group_t *resp)
                    tag_value->value.type < NEU_TYPE_ARRAY_STRING) {
             free(tag_value->value.value.bools.bools);
         }
+
+        if (tag_value->metas != NULL) {
+            for (int i = 0; i < tag_value->n_meta; i++) {
+                neu_free_dvalue(&tag_value->metas[i].value);
+            }
+            free(tag_value->metas);
+        }
     }
     free(resp->driver);
     free(resp->group);
@@ -1006,6 +1022,13 @@ neu_resp_read_paginate_free(neu_resp_read_group_paginate_t *resp)
                    tag_value->value.type < NEU_TYPE_ARRAY_STRING) {
             free(tag_value->value.value.bools.bools);
         }
+
+        if (tag_value->metas != NULL) {
+            for (int i = 0; i < tag_value->n_meta; i++) {
+                neu_free_dvalue(&tag_value->metas[i].value);
+            }
+            free(tag_value->metas);
+        }
     }
     free(resp->driver);
     free(resp->group);
@@ -1083,6 +1106,13 @@ static inline void neu_trans_data_free(neu_reqresp_trans_data_t *data)
                        tag_value->value.type < NEU_TYPE_ARRAY_STRING) {
                 free(tag_value->value.value.bools.bools);
             }
+
+            if (tag_value->metas != NULL) {
+                for (int i = 0; i < tag_value->n_meta; i++) {
+                    neu_free_dvalue(&tag_value->metas[i].value);
+                }
+                free(tag_value->metas);
+            }
         }
         utarray_free(data->tags);
         free(data->group);
@@ -1101,18 +1131,21 @@ static inline void neu_tag_value_to_json(neu_resp_tag_value_meta_t *tag_value,
     tag_json->name  = tag_value->tag;
     tag_json->error = 0;
 
-    for (int k = 0; k < NEU_TAG_META_SIZE; k++) {
-        if (strlen(tag_value->metas[k].name) > 0) {
-            tag_json->n_meta++;
-        } else {
-            break;
-        }
-    }
+    // for (int k = 0; k < NEU_TAG_META_SIZE; k++) {
+    //     if (strlen(tag_value->metas[k].name) > 0) {
+    //         tag_json->n_meta++;
+    //     } else {
+    //         break;
+    //     }
+    // }
+
+    tag_json->n_meta = tag_value->n_meta;
+
     if (tag_json->n_meta > 0) {
         tag_json->metas = (neu_json_tag_meta_t *) calloc(
             tag_json->n_meta, sizeof(neu_json_tag_meta_t));
     }
-    neu_json_metas_to_json(tag_value->metas, NEU_TAG_META_SIZE, tag_json);
+    neu_json_metas_to_json(tag_value->metas, tag_value->n_meta, tag_json);
 
     tag_json->datatag.bias = tag_value->datatag.bias;
 
@@ -1310,18 +1343,21 @@ neu_tag_value_to_json_paginate(neu_resp_tag_value_meta_paginate_t *tag_value,
     memcpy(tag_json->datatag.meta, tag_value->datatag.meta,
            NEU_TAG_META_LENGTH);
 
-    for (int k = 0; k < NEU_TAG_META_SIZE; k++) {
-        if (strlen(tag_value->metas[k].name) > 0) {
-            tag_json->n_meta++;
-        } else {
-            break;
-        }
-    }
+    // for (int k = 0; k < NEU_TAG_META_SIZE; k++) {
+    //     if (strlen(tag_value->metas[k].name) > 0) {
+    //         tag_json->n_meta++;
+    //     } else {
+    //         break;
+    //     }
+    // }
+
+    tag_json->n_meta = tag_value->n_meta;
+
     if (tag_json->n_meta > 0) {
         tag_json->metas = (neu_json_tag_meta_t *) calloc(
             tag_json->n_meta, sizeof(neu_json_tag_meta_t));
     }
-    neu_json_metas_to_json_paginate(tag_value->metas, NEU_TAG_META_SIZE,
+    neu_json_metas_to_json_paginate(tag_value->metas, tag_value->n_meta,
                                     tag_json);
 
     switch (tag_value->value.type) {

include/neuron/otel/otel_manager.h

@@ -42,6 +42,8 @@ neu_otel_trace_ctx neu_otel_find_trace2(void *req_ctx);
 void neu_otel_trace_set_final(neu_otel_trace_ctx ctx);
 
 void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num);
+void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
+                                             uint32_t           num);
 uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx);
 
 neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id);

include/neuron/tag.h

@@ -43,6 +43,10 @@ typedef enum {
     NEU_DATATAG_ENDIAN_BL32 = 5, // #BL 2,1,4,3
     NEU_DATATAG_ENDIAN_L64  = 6, // #L  8,7,6,5,4,3,2,1
     NEU_DATATAG_ENDIAN_B64  = 7, // #B  1,2,3,4,5,6,7,8
+    NEU_DATATAG_ENDIAN_LL64 = 8,
+    NEU_DATATAG_ENDIAN_LB64 = 9,
+    NEU_DATATAG_ENDIAN_BB64 = 10,
+    NEU_DATATAG_ENDIAN_BL64 = 11,
 } neu_datatag_endian_e;
 
 typedef enum {
@@ -62,6 +66,7 @@ typedef union {
     } value32;
     struct {
         neu_datatag_endian_e endian;
+        bool                 is_default;
     } value64;
     struct {
         uint16_t                  length;

include/neuron/type.h

@@ -890,6 +890,42 @@ static inline uint64_t neu_htonll(uint64_t u)
     return ret;
 }
 
+static inline uint64_t neu_htonlb(uint64_t u)
+{
+    uint8_t *in        = (uint8_t *) &u;
+    uint64_t out       = 0;
+    uint8_t *out_bytes = (uint8_t *) &out;
+
+    out_bytes[0] = in[1];
+    out_bytes[1] = in[0];
+    out_bytes[2] = in[3];
+    out_bytes[3] = in[2];
+    out_bytes[4] = in[5];
+    out_bytes[5] = in[4];
+    out_bytes[6] = in[7];
+    out_bytes[7] = in[6];
+
+    return out;
+}
+
+static inline uint64_t neu_htonbl(uint64_t u)
+{
+    uint8_t *in        = (uint8_t *) &u;
+    uint64_t out       = 0;
+    uint8_t *out_bytes = (uint8_t *) &out;
+
+    out_bytes[0] = in[6];
+    out_bytes[1] = in[7];
+    out_bytes[2] = in[4];
+    out_bytes[3] = in[5];
+    out_bytes[4] = in[2];
+    out_bytes[5] = in[3];
+    out_bytes[6] = in[0];
+    out_bytes[7] = in[1];
+
+    return out;
+}
+
 static inline uint64_t neu_ntohll(uint64_t u)
 {
     return neu_htonll(u);

plugins/datalayers/datalayers_handle.c

@@ -185,21 +185,27 @@ void db_write_task_consumer(neu_plugin_t *plugin)
     db_write_task_t *task = NULL;
 
     while (1) {
-        pthread_rwlock_rdlock(&plugin->plugin_mutex);
-        if (plugin->consumer_thread_stop_flag) {
-            pthread_rwlock_unlock(&plugin->plugin_mutex);
+        pthread_mutex_lock(&plugin->queue_mutex);
+
+        while (plugin->task_queue.size == 0 &&
+               !plugin->consumer_thread_stop_flag) {
+            pthread_cond_wait(&plugin->queue_not_empty, &plugin->queue_mutex);
+        }
+
+        if (plugin->consumer_thread_stop_flag && plugin->task_queue.size == 0) {
+            pthread_mutex_unlock(&plugin->queue_mutex);
             break;
         }
-        pthread_rwlock_unlock(&plugin->plugin_mutex);
 
         task = task_queue_pop(plugin, &plugin->task_queue);
+        pthread_mutex_unlock(&plugin->queue_mutex);
+
         if (task) {
             db_write_task_cb(task, plugin);
             task_free(task);
-        } else {
-            usleep(1000);
         }
     }
+
     plog_notice(plugin, "Consumer thread exiting gracefully.\n");
 }
 
@@ -558,10 +564,13 @@ int handle_trans_data(neu_plugin_t *            plugin,
     task->bool_tags       = bool_tags;
     task->string_tags     = string_tags;
 
-    task_queue_push(plugin, task);
-
     pthread_rwlock_unlock(&plugin->plugin_mutex);
 
+    pthread_mutex_lock(&plugin->queue_mutex);
+    task_queue_push(plugin, task);
+    pthread_cond_signal(&plugin->queue_not_empty);
+    pthread_mutex_unlock(&plugin->queue_mutex);
+
     return rv;
 }
 

plugins/datalayers/datalayers_plugin.h

@@ -68,6 +68,9 @@ struct neu_plugin {
     task_queue_t task_queue;
     pthread_t    consumer_thread;
 
+    pthread_mutex_t queue_mutex;
+    pthread_cond_t  queue_not_empty;
+
     pthread_rwlock_t plugin_mutex;
     bool             consumer_thread_stop_flag;
 

plugins/datalayers/datalayers_plugin_intf.c

@@ -33,9 +33,9 @@ extern const neu_plugin_module_t neu_plugin_module;
 
 void create_consumer_thread(neu_plugin_t *plugin)
 {
-    pthread_rwlock_wrlock(&plugin->plugin_mutex);
+    pthread_mutex_lock(&plugin->queue_mutex);
     plugin->consumer_thread_stop_flag = false;
-    pthread_rwlock_unlock(&plugin->plugin_mutex);
+    pthread_mutex_unlock(&plugin->queue_mutex);
 
     int status = pthread_create(&plugin->consumer_thread, NULL,
                                 (void *) db_write_task_consumer, plugin);
@@ -54,9 +54,10 @@ neu_plugin_t *datalayers_plugin_open(void)
 
 void stop_consumer_thread(pthread_t *consumer_thread, neu_plugin_t *plugin)
 {
-    pthread_rwlock_wrlock(&plugin->plugin_mutex);
+    pthread_mutex_lock(&plugin->queue_mutex);
     plugin->consumer_thread_stop_flag = true;
-    pthread_rwlock_unlock(&plugin->plugin_mutex);
+    pthread_cond_broadcast(&plugin->queue_not_empty);
+    pthread_mutex_unlock(&plugin->queue_mutex);
 
     int join_status = pthread_join(*consumer_thread, NULL);
     if (join_status != 0) {
@@ -81,11 +82,17 @@ int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
 
     pthread_rwlock_init(&plugin->plugin_mutex, NULL);
 
+    pthread_mutex_init(&plugin->queue_mutex, NULL);
+    pthread_cond_init(&plugin->queue_not_empty, NULL);
+
+    pthread_mutex_lock(&plugin->queue_mutex);
+    plugin->consumer_thread_stop_flag = false;
+    pthread_mutex_unlock(&plugin->queue_mutex);
+
     task_queue_init(&plugin->task_queue);
 
     pthread_create(&plugin->consumer_thread, NULL,
                    (void *) db_write_task_consumer, plugin);
-    pthread_detach(plugin->consumer_thread);
 
     NEU_PLUGIN_REGISTER_CACHED_QUEUE_SIZE_METRIC(plugin);
     NEU_PLUGIN_REGISTER_MAX_CACHED_QUEUE_SIZE_METRIC(plugin);
@@ -101,7 +108,11 @@ int datalayers_plugin_uninit(neu_plugin_t *plugin)
     pthread_rwlock_wrlock(&plugin->plugin_mutex);
 
     stop_consumer_thread(&plugin->consumer_thread, plugin);
+    pthread_mutex_lock(&plugin->queue_mutex);
     tasks_free(&plugin->task_queue);
+    pthread_mutex_unlock(&plugin->queue_mutex);
+    pthread_mutex_destroy(&plugin->queue_mutex);
+    pthread_cond_destroy(&plugin->queue_not_empty);
     datalayers_config_fini(&plugin->config);
 
     route_tbl_free(plugin->route_tbl);
@@ -135,10 +146,10 @@ static int config_datalayers_client(neu_plugin_t *             plugin,
     if (NULL == plugin->client) {
         plog_error(plugin, "datalayers client_create failed");
         plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
-        return NEU_ERR_DATALAYERS_CONNECT_FAILURE;
+    } else {
+        plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
     }
 
-    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
     return NEU_ERR_SUCCESS;
 }
 
@@ -149,7 +160,9 @@ int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting)
     datalayers_config_t config      = { 0 };
     if (plugin->client != NULL) {
         stop_consumer_thread(&plugin->consumer_thread, plugin);
+        pthread_mutex_lock(&plugin->queue_mutex);
         tasks_free(&plugin->task_queue);
+        pthread_mutex_unlock(&plugin->queue_mutex);
         client_destroy(plugin->client);
         plugin->client = NULL;
         create_consumer_thread(plugin);
@@ -185,7 +198,8 @@ int datalayers_plugin_start(neu_plugin_t *plugin)
         int ret = config_datalayers_client(plugin, &plugin->config);
         if (ret != 0) {
             plog_error(plugin, "datalayers started failed");
-            return NEU_ERR_DATALAYERS_CONNECT_FAILURE;
+            plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
+            return NEU_ERR_SUCCESS;
         }
     }
 

plugins/modbus/modbus-rtu.json

@@ -18,11 +18,11 @@
 		},
 		{
 			"type": 7,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 8,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 9,
@@ -30,7 +30,7 @@
 		},
 		{
 			"type": 10,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 11,
@@ -158,8 +158,8 @@
 		}
 	},
 	"endianess": {
-		"name": "Endianess",
-		"name_zh": "字节序",
+		"name": "Endianess of 4-Byte Data",
+		"name_zh": "4 字节数据字节序",
 		"description": "Tag byte order, ABCD corresponds to 1234",
 		"description_zh": "点位字节序，ABCD 对应 1234",
 		"attribute": "required",
@@ -186,6 +186,35 @@
 			]
 		}
 	},
+	"endianess_64": {
+		"name": "Endianess of 8-Byte Data",
+		"name_zh": "8 字节数据字节序",
+		"description": "Tag byte order, applicable to int64, uint64 and double types",
+		"description_zh": "点位字节序，适用于 int64，uint64 和 double 类型",
+		"attribute": "required",
+		"type": "map",
+		"default": 1,
+		"valid": {
+			"map": [
+				{
+					"key": "12 34 56 78",
+					"value": 1
+				},
+				{
+					"key": "21 43 65 87",
+					"value": 2
+				},
+				{
+					"key": "87 65 43 21",
+					"value": 3
+				},
+				{
+					"key": "78 56 34 12",
+					"value": 4
+				}
+			]
+		}
+	},
 	"address_base": {
 		"name": "Start Address",
 		"name_zh": "开始地址",

plugins/modbus/modbus-tcp.json

@@ -18,11 +18,11 @@
 		},
 		{
 			"type": 7,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 8,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 9,
@@ -30,7 +30,7 @@
 		},
 		{
 			"type": 10,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 11,
@@ -166,8 +166,8 @@
 		}
 	},
 	"endianess": {
-		"name": "Endianess",
-		"name_zh": "字节序",
+		"name": "Endianess of 4-Byte Data",
+		"name_zh": "4 字节数据字节序",
 		"description": "Tag byte order, ABCD corresponds to 1234",
 		"description_zh": "点位字节序，ABCD 对应 1234",
 		"attribute": "required",
@@ -194,6 +194,35 @@
 			]
 		}
 	},
+	"endianess_64": {
+		"name": "Endianess of 8-Byte Data",
+		"name_zh": "8 字节数据字节序",
+		"description": "Tag byte order, applicable to int64, uint64 and double types",
+		"description_zh": "点位字节序，适用于 int64，uint64 和 double 类型",
+		"attribute": "required",
+		"type": "map",
+		"default": 1,
+		"valid": {
+			"map": [
+				{
+					"key": "12 34 56 78",
+					"value": 1
+				},
+				{
+					"key": "21 43 65 87",
+					"value": 2
+				},
+				{
+					"key": "87 65 43 21",
+					"value": 3
+				},
+				{
+					"key": "78 56 34 12",
+					"value": 4
+				}
+			]
+		}
+	},
 	"address_base": {
 		"name": "Start Address",
 		"name_zh": "开始地址",

plugins/modbus/modbus.h

@@ -73,6 +73,13 @@ typedef enum modbus_endianess {
     MODBUS_CDAB = 4,
 } modbus_endianess;
 
+typedef enum modbus_endianess_64 {
+    MODBUS_LL = 1, // 12 34 56 78
+    MODBUS_LB = 2, // 21 43 65 87
+    MODBUS_BB = 3, // 87 65 43 21
+    MODBUS_BL = 4, // 78 56 34 12
+} modbus_endianess_64;
+
 typedef enum modbus_address_base {
     base_0 = 0,
     base_1 = 1,

plugins/modbus/modbus_point.c

@@ -238,7 +238,8 @@ modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte)
 }
 
 int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
-               modbus_endianess endianess, bool default_tag_endian)
+               modbus_endianess endianess, bool default_tag_endian,
+               modbus_endianess_64 endianess_64, bool default_tag_endian_64)
 {
     int n = 0;
     switch (type) {
@@ -260,6 +261,9 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
     case NEU_TYPE_DOUBLE:
     case NEU_TYPE_INT64:
     case NEU_TYPE_UINT64:
+        if (default_tag_endian_64) {
+            modbus_convert_endianess_64(value, endianess_64);
+        }
         n          = sizeof(uint64_t);
         value->u64 = neu_htonll(value->u64);
         break;
@@ -293,8 +297,9 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
     return n;
 }
 
-modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *       tags,
-                                                modbus_endianess endianess)
+modbus_write_cmd_sort_t *
+modbus_write_tags_sort(UT_array *tags, modbus_endianess endianess,
+                       modbus_endianess_64 endianess_64)
 {
     neu_tag_sort_result_t *result =
         neu_tag_sort(tags, tag_sort_write, tag_cmp_write);
@@ -316,17 +321,18 @@ modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *       tags,
         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, endianess, true);
+                n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
+                                        (*tag_s)->point.option, endianess, true,
+                                        endianess_64, true);
                 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, endianess,
-                               (*tag_s)->point.option.value32.is_default);
+                n_byte_tag = cal_n_byte(
+                    (*tag_s)->point.type, &(*tag_s)->value,
+                    (*tag_s)->point.option, endianess,
+                    (*tag_s)->point.option.value32.is_default, endianess_64,
+                    (*tag_s)->point.option.value64.is_default);
                 memcpy(sort_result->cmd[i].bytes +
                            2 *
                                ((*tag_s)->point.start_address -
@@ -558,4 +564,24 @@ void modbus_convert_endianess(neu_value_u *value, modbus_endianess endianess)
     default:
         break;
     }
+}
+
+void modbus_convert_endianess_64(neu_value_u *       value,
+                                 modbus_endianess_64 endianess_64)
+{
+    switch (endianess_64) {
+    case MODBUS_LL:
+        break;
+    case MODBUS_BB:
+        value->u64 = neu_htonll(value->u64);
+        break;
+    case MODBUS_LB:
+        value->u64 = neu_htonlb(value->u64);
+        break;
+    case MODBUS_BL:
+        value->u64 = neu_htonbl(value->u64);
+        break;
+    default:
+        break;
+    }
 }

plugins/modbus/modbus_point.h

@@ -81,12 +81,15 @@ typedef struct modbus_write_cmd_sort {
     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,
-                                                modbus_endianess endianess);
-void                     modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
+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, modbus_endianess endianess,
+                            modbus_endianess_64 endianess_64);
+void modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
 
 void modbus_convert_endianess(neu_value_u *value, modbus_endianess endianess);
+void modbus_convert_endianess_64(neu_value_u *       value,
+                                 modbus_endianess_64 endianess_64);
 
 #ifdef __cplusplus
 }

plugins/modbus/modbus_req.c

@@ -472,6 +472,10 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
             case NEU_TYPE_DOUBLE:
             case NEU_TYPE_INT64:
             case NEU_TYPE_UINT64:
+                if ((*p_tag)->option.value64.is_default) {
+                    modbus_convert_endianess_64(&dvalue.value,
+                                                plugin->endianess_64);
+                }
                 dvalue.value.u64 = neu_ntohll(dvalue.value.u64);
                 break;
             case NEU_TYPE_BIT: {
@@ -729,6 +733,9 @@ static uint8_t convert_value(neu_plugin_t *plugin, neu_value_u *value,
     case NEU_TYPE_DOUBLE:
     case NEU_TYPE_INT64:
     case NEU_TYPE_UINT64:
+        if (tag->option.value64.is_default) {
+            modbus_convert_endianess_64(value, plugin->endianess_64);
+        }
         value->u64 = neu_htonll(value->u64);
         n_byte     = sizeof(uint64_t);
         break;
@@ -821,7 +828,8 @@ int modbus_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
 
         utarray_push_back(gtags->tags, &p);
     }
-    gtags->cmd_sort = modbus_write_tags_sort(gtags->tags, plugin->endianess);
+    gtags->cmd_sort = modbus_write_tags_sort(gtags->tags, plugin->endianess,
+                                             plugin->endianess_64);
     for (uint16_t i = 0; i < gtags->cmd_sort->n_cmd; i++) {
         ret = write_modbus_points(plugin, &gtags->cmd_sort->cmd[i], req);
         if (ret <= 0) {

plugins/modbus/modbus_req.h

@@ -40,6 +40,7 @@ struct neu_plugin {
 
     modbus_protocol_e   protocol;
     modbus_endianess    endianess;
+    modbus_endianess_64 endianess_64;
     modbus_address_base address_base;
 
     uint16_t interval;

plugins/modbus/modbus_rtu.c

@@ -183,6 +183,9 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     neu_json_elem_t address_base = { .name = "address_base",
                                      .t    = NEU_JSON_INT };
 
+    neu_json_elem_t endianess_64 = { .name = "endianess_64",
+                                     .t    = NEU_JSON_INT };
+
     ret = neu_parse_param((char *) config, &err_param, 3, &link, &timeout,
                           &interval);
     if (ret != 0) {
@@ -244,6 +247,12 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
         address_base.v.val_int = base_1;
     }
 
+    ret = neu_parse_param((char *) config, &err_param, 1, &endianess_64);
+    if (ret != 0) {
+        free(err_param);
+        endianess_64.v.val_int = MODBUS_LL;
+    }
+
     param.log              = plugin->common.log;
     plugin->max_retries    = max_retries.v.val_int;
     plugin->retry_interval = retry_interval.v.val_int;
@@ -252,6 +261,7 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     plugin->degrade_time   = degrade_time.v.val_int;
     plugin->endianess      = endianess.v.val_int;
     plugin->address_base   = address_base.v.val_int;
+    plugin->endianess_64   = endianess_64.v.val_int;
 
     if (link.v.val_int == 0) {
         param.type = NEU_CONN_TTY_CLIENT;

plugins/modbus/modbus_tcp.c

@@ -190,6 +190,8 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     neu_json_elem_t endianess    = { .name = "endianess", .t = NEU_JSON_INT };
     neu_json_elem_t address_base = { .name = "address_base",
                                      .t    = NEU_JSON_INT };
+    neu_json_elem_t endianess_64 = { .name = "endianess_64",
+                                     .t    = NEU_JSON_INT };
 
     neu_json_elem_t  backup_ip   = { .name      = "backup_host",
                                   .t         = NEU_JSON_STR,
@@ -260,6 +262,12 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
         backup = true;
     }
 
+    ret = neu_parse_param((char *) config, &err_param, 1, &endianess_64);
+    if (ret != 0) {
+        free(err_param);
+        endianess_64.v.val_int = MODBUS_LL;
+    }
+
     param.log              = plugin->common.log;
     param_backup.log       = plugin->common.log;
     plugin->interval       = interval.v.val_int;
@@ -270,6 +278,7 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     plugin->degrade_cycle  = degrade_cycle.v.val_int;
     plugin->degrade_time   = degrade_time.v.val_int;
     plugin->endianess      = endianess.v.val_int;
+    plugin->endianess_64   = endianess_64.v.val_int;
     plugin->address_base   = address_base.v.val_int;
 
     if (mode.v.val_int == 1) {

plugins/mqtt/azure_iot_plugin.c

@@ -184,6 +184,7 @@ static int azure_parse_config(neu_plugin_t *plugin, const char *setting,
     config->cert                = cert.v.val_str;
     config->key                 = key.v.val_str;
     config->upload_drv_state    = false;
+    config->enable_topic        = true;
 
     plog_notice(plugin, "config MQTT version    : %d", config->version);
     plog_notice(plugin, "config client-id       : %s", config->client_id);

plugins/mqtt/mqtt.json

@@ -109,6 +109,16 @@
 		"default": true,
 		"valid": {}
 	},
+	"enable_topic": {
+		"name": "Enable driver topic",
+		"name_zh": "启动驱动相关主题",
+		"description": "Subscription driver related topics, including read/req, write/req and action/req.",
+		"description_zh": "是否订阅驱动相关主题，包括 read/req、write/req 和 action/req。",
+		"attribute": "optional",
+		"type": "bool",
+		"default": true,
+		"valid": {}
+	},
 	"write-req-topic": {
 		"name": "Write Request Topic",
 		"name_zh": "写请求主题",
@@ -116,7 +126,7 @@
 		"description_zh": "接收点位写入请求的 MQTT 主题。",
 		"attribute": "required",
 		"type": "string",
-		"default": "/neuron/${random_str}/write/req",
+		"default": "neuron/${random_str}/write/req",
 		"valid": {
 			"length": 255
 		}
@@ -128,7 +138,7 @@
 		"description_zh": "发送点位写入响应的 MQTT 主题。",
 		"attribute": "required",
 		"type": "string",
-		"default": "/neuron/${random_str}/write/resp",
+		"default": "neuron/${random_str}/write/resp",
 		"valid": {
 			"length": 255
 		}
@@ -139,7 +149,7 @@
 		"description": "MQTT topic prefix for driver function.",
 		"attribute": "required",
 		"type": "string",
-		"default": "/neuron/${random_str}",
+		"default": "neuron/${random_str}",
 		"valid": {
 			"length": 128
 		}
@@ -163,7 +173,7 @@
 			"field": "upload_drv_state",
 			"value": true
 		},
-		"default": "/neuron/${random_str}/state/update",
+		"default": "neuron/${random_str}/state/update",
 		"valid": {
 			"length": 255
 		}

plugins/mqtt/mqtt_config.c

@@ -291,13 +291,19 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
                                             .t    = NEU_JSON_INT };
     neu_json_elem_t host                = { .name = "host", .t = NEU_JSON_STR };
     neu_json_elem_t port                = { .name = "port", .t = NEU_JSON_INT };
-    neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR };
-    neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR };
-    neu_json_elem_t ssl      = { .name = "ssl", .t = NEU_JSON_BOOL };
-    neu_json_elem_t ca       = { .name = "ca", .t = NEU_JSON_STR };
-    neu_json_elem_t cert     = { .name = "cert", .t = NEU_JSON_STR };
-    neu_json_elem_t key      = { .name = "key", .t = NEU_JSON_STR };
-    neu_json_elem_t keypass  = { .name = "keypass", .t = NEU_JSON_STR };
+    neu_json_elem_t username     = { .name = "username", .t = NEU_JSON_STR };
+    neu_json_elem_t password     = { .name = "password", .t = NEU_JSON_STR };
+    neu_json_elem_t ssl          = { .name = "ssl", .t = NEU_JSON_BOOL };
+    neu_json_elem_t ca           = { .name = "ca", .t = NEU_JSON_STR };
+    neu_json_elem_t cert         = { .name = "cert", .t = NEU_JSON_STR };
+    neu_json_elem_t key          = { .name = "key", .t = NEU_JSON_STR };
+    neu_json_elem_t keypass      = { .name = "keypass", .t = NEU_JSON_STR };
+    neu_json_elem_t enable_topic = {
+        .name       = "enable_topic",
+        .t          = NEU_JSON_BOOL,
+        .v.val_bool = true,
+        .attribute  = NEU_JSON_ATTRIBUTE_OPTIONAL,
+    };
     neu_json_elem_t upload_drv_state = {
         .name       = "upload_drv_state",
         .t          = NEU_JSON_BOOL,
@@ -374,6 +380,11 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         free(schema.v.val_str);
     }
 
+    ret = neu_parse_param(setting, &err_param, 1, &enable_topic);
+    if (0 != ret) {
+        enable_topic.v.val_bool = true; // default to true
+    }
+
     ret = neu_parse_param(setting, &err_param, 1, &driver_topic_prefix);
     if (0 != ret) {
         plog_warn(plugin, "parsing driver topic prefix fail, key: `%s`",
@@ -456,6 +467,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     config->client_id           = client_id.v.val_str;
     config->qos                 = qos.v.val_int;
     config->format              = format.v.val_int;
+    config->enable_topic        = enable_topic.v.val_bool;
     config->write_req_topic     = write_req_topic.v.val_str;
     config->write_resp_topic    = write_resp_topic.v.val_str;
     config->cache               = offline_cache.v.val_bool;
@@ -483,6 +495,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     plog_notice(plugin, "config qos             : %d", config->qos);
     plog_notice(plugin, "config format          : %s",
                 mqtt_upload_format_str(config->format));
+    plog_notice(plugin, "config enable-topic    : %d", config->enable_topic);
     plog_notice(plugin, "config write-req-topic : %s", config->write_req_topic);
     plog_notice(plugin, "config write-resp-topic: %s",
                 config->write_resp_topic);

plugins/mqtt/mqtt_config.h

@@ -94,13 +94,14 @@ typedef struct {
 } mqtt_driver_topic_t;
 
 typedef struct {
-    neu_mqtt_version_e   version;          // mqtt version
-    char *               client_id;        // client id
-    neu_mqtt_qos_e       qos;              // message QoS
-    mqtt_upload_format_e format;           // upload format
-    char *               write_req_topic;  // write request topic
-    char *               write_resp_topic; // write response topic
+    neu_mqtt_version_e   version;   // mqtt version
+    char *               client_id; // client id
+    neu_mqtt_qos_e       qos;       // message QoS
+    mqtt_upload_format_e format;    // upload format
 
+    bool                enable_topic;        // default true
+    char *              write_req_topic;     // write request topic
+    char *              write_resp_topic;    // write response topic
     char *              driver_topic_prefix; // driver topic prefix
     mqtt_driver_topic_t driver_topic;
 
@@ -123,7 +124,6 @@ typedef struct {
     char *   keypass;             // client key password
                                   // remove in 2.6, keep it here
                                   // for backward compatibility
-
     size_t            n_schema_vt;
     mqtt_schema_vt_t *schema_vts;
 } mqtt_config_t;

plugins/mqtt/mqtt_handle.c

@@ -43,15 +43,28 @@ static void to_traceparent(uint8_t *trace_id, char *span_id, char *out)
 }
 
 static int tag_values_to_json(UT_array *tags, mqtt_static_vt_t *s_tags,
-                              size_t n_s_tags, neu_json_read_resp_t *json)
+                              size_t n_s_tags, neu_json_read_resp_t *json,
+                              bool filter_error)
 {
-    int index = 0;
+    int index   = 0;
+    int n_valid = 0;
 
-    if (0 == utarray_len(tags)) {
+    if (filter_error) {
+        utarray_foreach(tags, neu_resp_tag_value_meta_t *, tag_value)
+        {
+            if (tag_value->value.type != NEU_TYPE_ERROR) {
+                n_valid += 1;
+            }
+        }
+    } else {
+        n_valid = utarray_len(tags);
+    }
+
+    if (n_valid == 0) {
         return 0;
     }
 
-    json->n_tag = utarray_len(tags) + n_s_tags;
+    json->n_tag = n_valid + n_s_tags;
     json->tags  = (neu_json_read_resp_tag_t *) calloc(
         json->n_tag, sizeof(neu_json_read_resp_tag_t));
     if (NULL == json->tags) {
@@ -60,8 +73,15 @@ static int tag_values_to_json(UT_array *tags, mqtt_static_vt_t *s_tags,
 
     utarray_foreach(tags, neu_resp_tag_value_meta_t *, tag_value)
     {
-        neu_tag_value_to_json(tag_value, &json->tags[index]);
-        index += 1;
+        if (filter_error) {
+            if (tag_value->value.type != NEU_TYPE_ERROR) {
+                neu_tag_value_to_json(tag_value, &json->tags[index]);
+                index += 1;
+            }
+        } else {
+            neu_tag_value_to_json(tag_value, &json->tags[index]);
+            index += 1;
+        }
     }
 
     if (s_tags != NULL) {
@@ -77,27 +97,6 @@ static int tag_values_to_json(UT_array *tags, mqtt_static_vt_t *s_tags,
     return 0;
 }
 
-void filter_error_tags(neu_reqresp_trans_data_t *data)
-{
-    if (!data || !data->tags) {
-        return;
-    }
-
-    UT_array *filtered_tags;
-    utarray_new(filtered_tags, neu_resp_tag_value_meta_icd());
-
-    neu_resp_tag_value_meta_t *tag_ptr = NULL;
-    while ((tag_ptr = (neu_resp_tag_value_meta_t *) utarray_next(data->tags,
-                                                                 tag_ptr))) {
-        if (tag_ptr->value.type != NEU_TYPE_ERROR) {
-            utarray_push_back(filtered_tags, tag_ptr);
-        }
-    }
-
-    utarray_free(data->tags);
-    data->tags = filtered_tags;
-}
-
 char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
                            mqtt_upload_format_e format, mqtt_schema_vt_t *vts,
                            size_t n_vts, mqtt_static_vt_t *s_tags,
@@ -109,29 +108,32 @@ char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
                                         .timestamp = global_timestamp };
     neu_json_read_resp_t     json     = { 0 };
 
-    if (!plugin->config.upload_err && skip != NULL) {
-        filter_error_tags(data);
-
-        if (utarray_len(data->tags) == 0) {
-            *skip = true;
-            return NULL;
-        }
-    }
-
     if (format == MQTT_UPLOAD_FORMAT_CUSTOM) {
-        if (0 != tag_values_to_json(data->tags, NULL, 0, &json)) {
+        if (0 !=
+            tag_values_to_json(data->tags, NULL, 0, &json,
+                               !plugin->config.upload_err)) {
             plog_error(plugin, "tag_values_to_json fail");
             return NULL;
         }
     } else {
-        if (0 != tag_values_to_json(data->tags, s_tags, n_s_tags, &json)) {
+        if (0 !=
+            tag_values_to_json(data->tags, s_tags, n_s_tags, &json,
+                               !plugin->config.upload_err)) {
             plog_error(plugin, "tag_values_to_json fail");
             return NULL;
         }
     }
 
-    int ret;
+    if (json.n_tag == 0) {
+        plog_warn(plugin, "driver:%s group:%s, no valid tags", data->driver,
+                  data->group);
+        if (skip != NULL) {
+            *skip = true;
+        }
+        return NULL;
+    }
 
+    int ret;
     switch (format) {
     case MQTT_UPLOAD_FORMAT_VALUES:
         neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header,
@@ -186,7 +188,7 @@ static char *generate_read_resp_json(neu_plugin_t *         plugin,
     char *               json_str = NULL;
     neu_json_read_resp_t json     = { 0 };
 
-    if (0 != tag_values_to_json(data->tags, NULL, 0, &json)) {
+    if (0 != tag_values_to_json(data->tags, NULL, 0, &json, false)) {
         plog_error(plugin, "tag_values_to_json fail");
         return NULL;
     }
@@ -1184,7 +1186,7 @@ int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                 break;
             }
 
-            for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
+            for (int i = 0; i < tag_value->n_meta; i++) {
                 if (strlen(tag_value->metas[i].name) > 0) {
                     if (strncmp(tag_value->metas[i].name, "q", 1) == 0) {
                         tag->has_q = true;
@@ -1414,7 +1416,7 @@ int handle_trans_data(neu_plugin_t *            plugin,
                     break;
                 }
 
-                for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
+                for (int i = 0; i < tag_value->n_meta; i++) {
                     if (strlen(tag_value->metas[i].name) > 0) {
                         if (strncmp(tag_value->metas[i].name, "q", 1) == 0) {
                             tag->has_q = true;

plugins/mqtt/mqtt_plugin_intf.c

@@ -265,13 +265,13 @@ static int create_topic(neu_plugin_t *plugin)
         return 0;
     }
 
-    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
+    neu_asprintf(&plugin->read_req_topic, "neuron/%s/read/req",
                  plugin->common.name);
     if (NULL == plugin->read_req_topic) {
         return -1;
     }
 
-    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
+    neu_asprintf(&plugin->read_resp_topic, "neuron/%s/read/resp",
                  plugin->common.name);
     if (NULL == plugin->read_resp_topic) {
         free(plugin->read_req_topic);
@@ -387,7 +387,6 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
     int           rv          = 0;
     const char *  plugin_name = neu_plugin_module.module_name;
     mqtt_config_t config      = { 0 };
-    bool          started     = false;
 
     rv = plugin->parse_config(plugin, setting, &config);
     if (0 != rv) {
@@ -395,33 +394,11 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
         return NEU_ERR_NODE_SETTING_INVALID;
     }
 
-    if (NULL == plugin->client) {
-        plugin->client = neu_mqtt_client_new(config.version);
-        if (NULL == plugin->client) {
-            plog_error(plugin, "neu_mqtt_client_new fail");
-            rv = NEU_ERR_EINTERNAL;
-            goto error;
-        }
-    } else if (neu_mqtt_client_is_open(plugin->client)) {
-        started = true;
-        plugin->unsubscribe(plugin, &plugin->config);
-        rv = neu_mqtt_client_close(plugin->client);
-        if (0 != rv) {
-            plog_error(plugin, "neu_mqtt_client_close fail");
-            rv = NEU_ERR_EINTERNAL;
-            goto error;
-        }
-        if (neu_mqtt_client_check_version_change(plugin->client,
-                                                 config.version)) {
-            neu_mqtt_client_free(plugin->client);
-            plugin->client = neu_mqtt_client_new(config.version);
-        }
-    } else if (neu_mqtt_client_check_version_change(plugin->client,
-                                                    config.version)) {
-        // plugin stopped and version changed
+    if (plugin->client != NULL) {
+        neu_mqtt_client_close(plugin->client);
         neu_mqtt_client_free(plugin->client);
-        plugin->client = neu_mqtt_client_new(config.version);
     }
+    plugin->client = neu_mqtt_client_new(config.version);
 
     rv = config_mqtt_client(plugin, plugin->client, &config);
     if (0 != rv) {
@@ -429,17 +406,17 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
         goto error;
     }
 
-    if (started) {
-        if (0 != neu_mqtt_client_open(plugin->client)) {
-            plog_error(plugin, "neu_mqtt_client_open fail");
-            rv = NEU_ERR_MQTT_CONNECT_FAILURE;
-            goto error;
-        }
-        if (0 != start_hearbeat_timer(plugin, config.heartbeat_interval)) {
-            plog_error(plugin, "start hearbeat_timer failed");
-            rv = NEU_ERR_EINTERNAL;
-            goto error;
-        }
+    if (0 != neu_mqtt_client_open(plugin->client)) {
+        plog_error(plugin, "neu_mqtt_client_open fail");
+        rv = NEU_ERR_MQTT_CONNECT_FAILURE;
+        goto error;
+    }
+    if (0 != start_hearbeat_timer(plugin, config.heartbeat_interval)) {
+        plog_error(plugin, "start hearbeat_timer failed");
+        rv = NEU_ERR_EINTERNAL;
+        goto error;
+    }
+    if (config.enable_topic) {
         if (0 != (rv = plugin->subscribe(plugin, &config))) {
             goto error;
         }
@@ -483,7 +460,10 @@ int mqtt_plugin_start(neu_plugin_t *plugin)
         goto end;
     }
 
-    rv = plugin->subscribe(plugin, &plugin->config);
+    rv = 0;
+    if (plugin->config.enable_topic) {
+        rv = plugin->subscribe(plugin, &plugin->config);
+    }
 
 end:
     if (0 == rv) {
@@ -499,7 +479,9 @@ end:
 int mqtt_plugin_stop(neu_plugin_t *plugin)
 {
     if (plugin->client) {
-        plugin->unsubscribe(plugin, &plugin->config);
+        if (plugin->config.enable_topic) {
+            plugin->unsubscribe(plugin, &plugin->config);
+        }
         neu_mqtt_client_close(plugin->client);
         plog_notice(plugin, "mqtt client closed");
     }

plugins/restful/datalayers_handle.c

@@ -41,27 +41,8 @@ void handle_datalayers_get_groups(nng_aio *aio)
 
     NEU_VALIDATE_JWT(aio);
 
-    // required parameter
     snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
 
-    // optional parameter
-    ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
-    if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
-            neu_http_response(aio, error_code.error, result_error);
-        })
-        return;
-    }
-
-    // optional parameter
-    ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
-    if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
-            neu_http_response(aio, error_code.error, result_error);
-        })
-        return;
-    }
-
     ret = neu_plugin_op(plugin, header, &cmd);
     if (ret != 0) {
         NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
@@ -111,7 +92,22 @@ void handle_datalayers_get_tags(nng_aio *aio)
 
     NEU_VALIDATE_JWT(aio);
 
-    // required parameter
+    ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
+    if (ret <= 0 || (size_t) ret == sizeof(cmd.driver)) {
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
+            neu_http_response(aio, error_code.error, result_error);
+        })
+        return;
+    }
+
+    ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
+    if (ret <= 0 || (size_t) ret == sizeof(cmd.group)) {
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
+            neu_http_response(aio, error_code.error, result_error);
+        })
+        return;
+    }
+
     snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
 
     ret = neu_plugin_op(plugin, header, &cmd);
@@ -137,6 +133,63 @@ void handle_datalayers_get_tags_resp(nng_aio *                       aio,
 
     neu_http_ok(aio, result);
 
+    free(result);
+    if (tags) {
+        utarray_foreach(tags->infos, neu_resp_get_sub_driver_tags_info_t *,
+                        info)
+        {
+            utarray_free(info->tags);
+        }
+        utarray_free(tags->infos);
+    }
+}
+
+void handle_datalayers_get_tag(nng_aio *aio)
+{
+    int                           ret    = 0;
+    neu_plugin_t *                plugin = neu_rest_get_plugin();
+    neu_req_get_sub_driver_tags_t cmd    = { 0 };
+    neu_reqresp_head_t            header = {
+        .ctx             = aio,
+        .type            = NEU_REQ_GET_DATALAYERS_TAG,
+        .otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM,
+    };
+
+    NEU_VALIDATE_JWT(aio);
+
+    ret = neu_http_get_param_str(aio, "tag", cmd.tag, sizeof(cmd.tag));
+    if (ret <= 0 || (size_t) ret == sizeof(cmd.tag)) {
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
+            neu_http_response(aio, error_code.error, result_error);
+        })
+        return;
+    }
+
+    snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
+
+    ret = neu_plugin_op(plugin, header, &cmd);
+    if (ret != 0) {
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
+            neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+        });
+    }
+}
+
+void handle_datalayers_get_tag_resp(nng_aio *                       aio,
+                                    neu_resp_get_sub_driver_tags_t *tags)
+{
+    char *result = NULL;
+    int   ret    = neu_json_encode_by_fn(
+        tags, neu_json_encode_datalayers_get_tag_resp, &result);
+
+    if (ret != 0 || result == NULL) {
+        neu_http_response(
+            aio, 500, "{\"error\":4200,\"reason\":\"failed to encode tags\"}");
+        return;
+    }
+
+    neu_http_ok(aio, result);
+
     free(result);
     if (tags) {
         utarray_foreach(tags->infos, neu_resp_get_sub_driver_tags_info_t *,

plugins/restful/datalayers_handle.h

@@ -32,4 +32,8 @@ void handle_datalayers_get_tags(nng_aio *aio);
 void handle_datalayers_get_tags_resp(nng_aio *                       aio,
                                      neu_resp_get_sub_driver_tags_t *tags);
 
+void handle_datalayers_get_tag(nng_aio *aio);
+void handle_datalayers_get_tag_resp(nng_aio *                       aio,
+                                    neu_resp_get_sub_driver_tags_t *tags);
+
 #endif

plugins/restful/handle.c

@@ -155,6 +155,9 @@ static struct neu_http_handler cors_handler[] = {
     {
         .url = "/api/v2/datalayers/tags",
     },
+    {
+        .url = "/api/v2/datalayers/tag",
+    },
 };
 
 static struct neu_http_handler rest_handlers[] = {
@@ -513,6 +516,12 @@ static struct neu_http_handler rest_handlers[] = {
         .url           = "/api/v2/datalayers/tags",
         .value.handler = handle_datalayers_get_tags,
     },
+    {
+        .method        = NEU_HTTP_METHOD_GET,
+        .type          = NEU_HTTP_HANDLER_FUNCTION,
+        .url           = "/api/v2/datalayers/tag",
+        .value.handler = handle_datalayers_get_tag,
+    },
 };
 
 void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)

plugins/restful/rest.c

@@ -298,6 +298,13 @@ static int dashb_plugin_request(neu_plugin_t *      plugin,
             neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
         }
         break;
+    case NEU_RESP_GET_DATALAYERS_TAG:
+        handle_datalayers_get_tag_resp(header->ctx,
+                                       (neu_resp_get_sub_driver_tags_t *) data);
+        if (neu_otel_control_is_started() && trace) {
+            neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
+        }
+        break;
     case NEU_RESP_GET_NODE_SETTING:
         handle_get_node_setting_resp(header->ctx,
                                      (neu_resp_get_node_setting_t *) data);

src/adapter/adapter.c

@@ -123,8 +123,19 @@ static void *adapter_consumer(void *arg)
         nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
                    adapter->name, header->sender, header->ctx,
                    neu_reqresp_type_string(header->type), n);
-        adapter->module->intf_funs->request(
-            adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        if (adapter->state == NEU_NODE_RUNNING_STATE_RUNNING) {
+            adapter->module->intf_funs->request(
+                adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        } else {
+            void *ctx = ((neu_reqresp_trans_data_t *) &header[1])->trace_ctx;
+            if (neu_otel_data_is_started() && ctx) {
+                neu_otel_trace_ctx trace = neu_otel_find_trace(ctx);
+                if (trace) {
+                    neu_otel_trace_reduce_expected_span_num(trace, 1);
+                }
+            }
+        }
+
         neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
         neu_msg_free(msg);
     }
@@ -757,6 +768,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_RESP_GET_SUBSCRIBE_GROUP:
     case NEU_RESP_GET_DATALAYERS_GROUPS:
     case NEU_RESP_GET_DATALAYERS_TAGS:
+    case NEU_RESP_GET_DATALAYERS_TAG:
     case NEU_RESP_ADD_TAG:
     case NEU_RESP_ADD_GTAG:
     case NEU_RESP_UPDATE_TAG:

src/adapter/driver/cache.c

@@ -45,7 +45,8 @@ struct elem {
     neu_dvalue_t value;
     neu_dvalue_t value_old;
 
-    neu_tag_meta_t metas[NEU_TAG_META_SIZE];
+    neu_tag_meta_t *metas;
+    int             n_meta;
 
     tkey_t         key;
     UT_hash_handle hh;
@@ -115,6 +116,13 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
             free(elem->value.value.bools.bools);
         }
 
+        if (elem->metas != NULL) {
+            for (int i = 0; i < elem->n_meta; i++) {
+                neu_free_dvalue(&elem->metas[i].value);
+            }
+            free(elem->metas);
+        }
+
         free(elem);
     }
 
@@ -764,9 +772,16 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
         }
         elem->value.type = value.type;
 
-        memset(elem->metas, 0, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE);
-        for (int i = 0; i < n_meta; i++) {
-            memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t));
+        if (metas != NULL) {
+            if (elem->metas != NULL) {
+                free(elem->metas);
+            }
+            elem->metas = calloc(n_meta, sizeof(neu_tag_meta_t));
+            for (int i = 0; i < n_meta; i++) {
+                memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t));
+            }
+
+            elem->n_meta = n_meta;
         }
     }
 
@@ -784,7 +799,7 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
 
 int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
                               const char *tag, neu_driver_cache_value_t *value,
-                              neu_tag_meta_t *metas, int n_meta)
+                              neu_tag_meta_t **metas, int *n_meta)
 {
     struct elem *elem = NULL;
     int          ret  = -1;
@@ -798,8 +813,14 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
         value->value.type      = elem->value.type;
         value->value.precision = elem->value.precision;
 
-        assert(n_meta <= NEU_TAG_META_SIZE);
-        memcpy(metas, elem->metas, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE);
+        // assert(n_meta <= NEU_TAG_META_SIZE);
+        if (elem->metas) {
+            *metas = calloc(elem->n_meta, sizeof(neu_tag_meta_t));
+            memcpy(*metas, elem->metas, sizeof(neu_tag_meta_t) * elem->n_meta);
+        } else {
+            *metas = NULL;
+        }
+        *n_meta = elem->n_meta;
 
         switch (elem->value.type) {
         case NEU_TYPE_INT8:
@@ -941,13 +962,6 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
             break;
         }
 
-        for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
-            if (strlen(elem->metas[i].name) > 0) {
-                memcpy(&value->metas[i], &elem->metas[i],
-                       sizeof(neu_tag_meta_t));
-            }
-        }
-
         ret = 0;
     }
 
@@ -959,7 +973,7 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
 int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
                                       const char *group, const char *tag,
                                       neu_driver_cache_value_t *value,
-                                      neu_tag_meta_t *metas, int n_meta)
+                                      neu_tag_meta_t **metas, int *n_meta)
 {
     struct elem *elem = NULL;
     int          ret  = -1;
@@ -973,8 +987,14 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
         value->value.type      = elem->value.type;
         value->value.precision = elem->value.precision;
 
-        assert(n_meta <= NEU_TAG_META_SIZE);
-        memcpy(metas, elem->metas, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE);
+        // assert(n_meta <= NEU_TAG_META_SIZE);
+        if (elem->metas) {
+            *metas = calloc(elem->n_meta, sizeof(neu_tag_meta_t));
+            memcpy(*metas, elem->metas, sizeof(neu_tag_meta_t) * elem->n_meta);
+        } else {
+            *metas = NULL;
+        }
+        *n_meta = elem->n_meta;
 
         switch (elem->value.type) {
         case NEU_TYPE_INT8:
@@ -1116,13 +1136,6 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
             break;
         }
 
-        for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
-            if (strlen(elem->metas[i].name) > 0) {
-                memcpy(&value->metas[i], &elem->metas[i],
-                       sizeof(neu_tag_meta_t));
-            }
-        }
-
         if (elem->value.type != NEU_TYPE_ERROR) {
             elem->changed = false;
         }
@@ -1165,6 +1178,12 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
                    elem->value.type < NEU_TYPE_ARRAY_STRING) {
             free(elem->value.value.bools.bools);
         }
+        if (elem->metas != NULL) {
+            for (int i = 0; i < elem->n_meta; i++) {
+                neu_free_dvalue(&elem->metas[i].value);
+            }
+            free(elem->metas);
+        }
         free(elem);
     }
 

src/adapter/driver/cache.h

@@ -50,17 +50,16 @@ void neu_driver_cache_update_trace(neu_driver_cache_t *cache, const char *group,
 void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group);
 
 typedef struct {
-    neu_dvalue_t   value;
-    int64_t        timestamp;
-    neu_tag_meta_t metas[NEU_TAG_META_SIZE];
+    neu_dvalue_t value;
+    int64_t      timestamp;
 } neu_driver_cache_value_t;
 
 int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
                               const char *tag, neu_driver_cache_value_t *value,
-                              neu_tag_meta_t *metas, int n_meta);
+                              neu_tag_meta_t **metas, int *n_meta);
 int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
                                       const char *group, const char *tag,
                                       neu_driver_cache_value_t *value,
-                                      neu_tag_meta_t *metas, int n_meta);
+                                      neu_tag_meta_t **metas, int *n_meta);
 
 #endif

src/adapter/driver/driver.c

@@ -947,9 +947,17 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
     case NEU_TYPE_LWORD:
         switch (tag->option.value64.endian) {
         case NEU_DATATAG_ENDIAN_B64:
+        case NEU_DATATAG_ENDIAN_BB64:
             value->value.u64 = neu_ntohll(value->value.u64);
             break;
+        case NEU_DATATAG_ENDIAN_LB64:
+            value->value.u64 = neu_htonlb(value->value.u64);
+            break;
+        case NEU_DATATAG_ENDIAN_BL64:
+            value->value.u64 = neu_htonbl(value->value.u64);
+            break;
         case NEU_DATATAG_ENDIAN_L64:
+        case NEU_DATATAG_ENDIAN_LL64:
         default:
             break;
         }
@@ -2238,6 +2246,7 @@ static int report_callback(void *usr_data)
     data->driver = strdup(group->driver->adapter.name);
     data->group  = strdup(group->name);
     utarray_new(data->tags, neu_resp_tag_value_meta_icd());
+    utarray_reserve(data->tags, utarray_len(tags));
 
     void *trace_ctx =
         neu_driver_cache_get_trace(group->driver->cache, group->name);
@@ -2289,20 +2298,6 @@ static int report_callback(void *usr_data)
 
             utarray_foreach(group->apps, sub_app_t *, app)
             {
-
-                neu_adapter_t *sub_app =
-                    neu_node_manager_find(g_manager->node_manager, app->app);
-                if (sub_app != NULL) {
-                    if (sub_app->state != NEU_NODE_RUNNING_STATE_RUNNING) {
-                        neu_trans_data_free(data);
-                        continue;
-                    }
-                } else {
-                    nlog_error("sub app %s not exist", app->app);
-                    neu_trans_data_free(data);
-                    continue;
-                }
-
                 if (group->driver->adapter.cb_funs.responseto(
                         &group->driver->adapter, &header, data, app->addr) !=
                     0) {
@@ -2343,6 +2338,13 @@ static int report_callback(void *usr_data)
                 } else {
                     neu_free_dvalue(&tag_value->value);
                 }
+
+                if (tag_value->metas != NULL) {
+                    for (int i = 0; i < tag_value->n_meta; i++) {
+                        neu_free_dvalue(&tag_value->metas[i].value);
+                    }
+                    free(tag_value->metas);
+                }
             }
             utarray_free(data->tags);
             free(data->group);
@@ -2547,15 +2549,15 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
 
         if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) {
             if (neu_driver_cache_meta_get_changed(cache, group, tag->name,
-                                                  &value, tag_value.metas,
-                                                  NEU_TAG_META_SIZE) != 0) {
+                                                  &value, &tag_value.metas,
+                                                  &tag_value.n_meta) != 0) {
                 nlog_debug("tag: %s not changed", tag->name);
                 continue;
             }
         } else {
             if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
-                                          tag_value.metas,
-                                          NEU_TAG_META_SIZE) != 0) {
+                                          &tag_value.metas,
+                                          &tag_value.n_meta) != 0) {
                 strcpy(tag_value.tag, tag->name);
                 tag_value.value.type      = NEU_TYPE_ERROR;
                 tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;
@@ -2628,9 +2630,17 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
         case NEU_TYPE_UINT64:
             switch (tag->option.value64.endian) {
             case NEU_DATATAG_ENDIAN_B64:
+            case NEU_DATATAG_ENDIAN_BB64:
                 value.value.value.u64 = neu_htonll(value.value.value.u64);
                 break;
+            case NEU_DATATAG_ENDIAN_LB64:
+                value.value.value.u64 = neu_htonlb(value.value.value.u64);
+                break;
+            case NEU_DATATAG_ENDIAN_BL64:
+                value.value.value.u64 = neu_htonbl(value.value.value.u64);
+                break;
             case NEU_DATATAG_ENDIAN_L64:
+            case NEU_DATATAG_ENDIAN_LL64:
             default:
                 break;
             }
@@ -2734,8 +2744,8 @@ static void read_group(int64_t timestamp, int64_t timeout,
         tag_value.datatag.bias = tag->bias;
 
         if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
-                                      tag_value.metas,
-                                      NEU_TAG_META_SIZE) != 0) {
+                                      &tag_value.metas,
+                                      &tag_value.n_meta) != 0) {
             tag_value.value.type      = NEU_TYPE_ERROR;
             tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;
 
@@ -2794,9 +2804,17 @@ static void read_group(int64_t timestamp, int64_t timeout,
         case NEU_TYPE_UINT64:
             switch (tag->option.value64.endian) {
             case NEU_DATATAG_ENDIAN_B64:
+            case NEU_DATATAG_ENDIAN_BB64:
                 value.value.value.u64 = neu_htonll(value.value.value.u64);
                 break;
+            case NEU_DATATAG_ENDIAN_LB64:
+                value.value.value.u64 = neu_htonlb(value.value.value.u64);
+                break;
+            case NEU_DATATAG_ENDIAN_BL64:
+                value.value.value.u64 = neu_htonbl(value.value.value.u64);
+                break;
             case NEU_DATATAG_ENDIAN_L64:
+            case NEU_DATATAG_ENDIAN_LL64:
             default:
                 break;
             }
@@ -2914,8 +2932,8 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
         }
 
         if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
-                                      tag_value.metas,
-                                      NEU_TAG_META_SIZE) != 0) {
+                                      &tag_value.metas,
+                                      &tag_value.n_meta) != 0) {
             tag_value.value.type      = NEU_TYPE_ERROR;
             tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;
 
@@ -2974,9 +2992,17 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
         case NEU_TYPE_UINT64:
             switch (tag->option.value64.endian) {
             case NEU_DATATAG_ENDIAN_B64:
+            case NEU_DATATAG_ENDIAN_BB64:
                 value.value.value.u64 = neu_htonll(value.value.value.u64);
                 break;
+            case NEU_DATATAG_ENDIAN_LB64:
+                value.value.value.u64 = neu_htonlb(value.value.value.u64);
+                break;
+            case NEU_DATATAG_ENDIAN_BL64:
+                value.value.value.u64 = neu_htonbl(value.value.value.u64);
+                break;
             case NEU_DATATAG_ENDIAN_L64:
+            case NEU_DATATAG_ENDIAN_LL64:
             default:
                 break;
             }

src/base/msg_internal.h

@@ -135,7 +135,9 @@ extern "C" {
     XX(NEU_REQ_GET_DATALAYERS_GROUPS, neu_req_get_subscribe_group_t)   \
     XX(NEU_RESP_GET_DATALAYERS_GROUPS, neu_resp_get_subscribe_group_t) \
     XX(NEU_REQ_GET_DATALAYERS_TAGS, neu_req_get_sub_driver_tags_t)     \
-    XX(NEU_RESP_GET_DATALAYERS_TAGS, neu_resp_get_sub_driver_tags_t)
+    XX(NEU_RESP_GET_DATALAYERS_TAGS, neu_resp_get_sub_driver_tags_t)   \
+    XX(NEU_REQ_GET_DATALAYERS_TAG, neu_req_get_sub_driver_tags_t)      \
+    XX(NEU_RESP_GET_DATALAYERS_TAG, neu_resp_get_sub_driver_tags_t)
 
 static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
 {

src/base/tag.c

@@ -262,21 +262,35 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
     case NEU_TYPE_UINT64: {
         char *op = find_last_character(datatag->address, '#');
 
-        option->value64.endian = NEU_DATATAG_ENDIAN_L64;
+        option->value64.endian     = NEU_DATATAG_ENDIAN_L64;
+        option->value64.is_default = true;
         if (op != NULL) {
-            char e = 0;
-            sscanf(op, "#%c", &e);
+            char e1 = 0;
+            char e2 = 0;
+            int  n  = sscanf(op, "#%c%c", &e1, &e2);
 
-            switch (e) {
-            case 'B':
-                option->value64.endian = NEU_DATATAG_ENDIAN_B64;
-                break;
-            case 'L':
-                option->value64.endian = NEU_DATATAG_ENDIAN_L64;
-                break;
-            default:
-                option->value64.endian = NEU_DATATAG_ENDIAN_L64;
-                break;
+            if (n == 2) {
+                if (e1 == 'B' && e2 == 'B') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_BB64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'B' && e2 == 'L') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_BL64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'L' && e2 == 'L') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_LL64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'L' && e2 == 'B') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_LB64;
+                    option->value64.is_default = false;
+                }
+            } else if (n == 1) {
+                if (e1 == 'B') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_B64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'L') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_L64;
+                    option->value64.is_default = false;
+                }
             }
         }
 

src/core/manager.c

@@ -1056,10 +1056,18 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             (neu_req_get_sub_driver_tags_t *) &header[1];
         neu_resp_get_sub_driver_tags_t resp = { 0 };
         UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
+        bool      found  = false;
 
         utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd());
         utarray_foreach(groups, neu_resp_subscribe_info_t *, info)
         {
+            if (strcmp(info->driver, cmd->driver) != 0 ||
+                strcmp(info->group, cmd->group) != 0) {
+                continue;
+            }
+
+            found = true;
+
             neu_resp_get_sub_driver_tags_info_t in = { 0 };
             neu_adapter_t *                     driver =
                 neu_node_manager_find(manager->node_manager, info->driver);
@@ -1074,12 +1082,82 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         utarray_free(groups);
 
+        if (!found) {
+            utarray_free(resp.infos);
+            neu_resp_error_t e = { .error = NEU_ERR_GROUP_NOT_EXIST };
+            header->type       = NEU_RESP_ERROR;
+            neu_msg_exchange(header);
+            reply(manager, header, &e);
+            break;
+        }
+
         strcpy(header->receiver, header->sender);
         header->type = NEU_RESP_GET_DATALAYERS_TAGS;
         reply(manager, header, &resp);
 
         break;
     }
+    case NEU_REQ_GET_DATALAYERS_TAG: {
+        neu_req_get_sub_driver_tags_t *cmd =
+            (neu_req_get_sub_driver_tags_t *) &header[1];
+        neu_resp_get_sub_driver_tags_t resp = { 0 };
+        UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
+        bool      found  = false;
+
+        utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd());
+
+        utarray_foreach(groups, neu_resp_subscribe_info_t *, info)
+        {
+            neu_resp_get_sub_driver_tags_info_t in = { 0 };
+            neu_adapter_t *                     driver =
+                neu_node_manager_find(manager->node_manager, info->driver);
+            assert(driver != NULL);
+
+            neu_adapter_driver_get_value_tag((neu_adapter_driver_t *) driver,
+                                             info->group, &in.tags);
+
+            if (in.tags) {
+                UT_array *filtered_tags = NULL;
+                utarray_new(filtered_tags, neu_tag_get_icd());
+
+                utarray_foreach(in.tags, neu_datatag_t *, tag)
+                {
+                    if (strstr(tag->name, cmd->tag) != NULL) {
+                        utarray_push_back(filtered_tags, tag);
+                    }
+                }
+
+                utarray_free(in.tags);
+
+                if (utarray_len(filtered_tags) > 0) {
+                    strcpy(in.driver, info->driver);
+                    strcpy(in.group, info->group);
+                    in.tags = filtered_tags;
+                    utarray_push_back(resp.infos, &in);
+                    found = true;
+                } else {
+                    utarray_free(filtered_tags);
+                }
+            }
+        }
+
+        utarray_free(groups);
+
+        if (!found) {
+            utarray_free(resp.infos);
+            neu_resp_error_t e = { .error = NEU_ERR_TAG_NOT_EXIST };
+            header->type       = NEU_RESP_ERROR;
+            neu_msg_exchange(header);
+            reply(manager, header, &e);
+            break;
+        }
+
+        strcpy(header->receiver, header->sender);
+        header->type = NEU_RESP_GET_DATALAYERS_TAG;
+        reply(manager, header, &resp);
+
+        break;
+    }
     case NEU_REQ_GET_NODES_STATE: {
         neu_resp_get_nodes_state_t resp = { 0 };
         UT_array *states = neu_node_manager_get_state(manager->node_manager);

src/core/storage.c

@@ -18,6 +18,7 @@
  **/
 #include "errcodes.h"
 #include "utils/log.h"
+#include "utils/time.h"
 
 #include "adapter/storage.h"
 #include "storage.h"
@@ -140,6 +141,8 @@ int manager_load_node(neu_manager_t *manager)
         return -1;
     }
 
+    int i = 0;
+
     utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
     {
         rv                    = neu_manager_add_node(manager, node_info->name,
@@ -149,6 +152,10 @@ int manager_load_node(neu_manager_t *manager)
         nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",
                     ok_or_err, node_info->type, node_info->name,
                     node_info->plugin_name, node_info->state);
+        i++;
+        if (i % 10 == 0) {
+            neu_msleep(10);
+        }
     }
 
     utarray_free(node_infos);
@@ -160,6 +167,8 @@ int manager_load_subscribe(neu_manager_t *manager)
     UT_array *nodes =
         neu_node_manager_get(manager->node_manager, NEU_NA_TYPE_APP);
 
+    int i = 0;
+
     utarray_foreach(nodes, neu_resp_node_info_t *, node)
     {
         int       rv        = 0;
@@ -186,6 +195,10 @@ int manager_load_subscribe(neu_manager_t *manager)
                                                info->group_name, app_port,
                                                info->params, info->static_tags);
                 }
+                i++;
+                if (i % 10 == 0) {
+                    neu_msleep(10);
+                }
             }
 
             utarray_free(sub_infos);

src/otel/otel_manager.c

@@ -540,6 +540,16 @@ void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num)
     trace_ctx->expected_span_num = num;
 }
 
+void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
+                                             uint32_t           num)
+{
+    trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
+    trace_ctx->expected_span_num -= num;
+    if (trace_ctx->expected_span_num <= 0) {
+        trace_ctx->final = true;
+    }
+}
+
 uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx)
 {
     trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;

src/parser/neu_json_datalayers.c

@@ -163,6 +163,7 @@ int neu_json_encode_datalayers_get_tags_resp(void *json_object_param,
                 json_object_set_new(tag_obj, "name", json_string(tag->name));
                 json_object_set_new(tag_obj, "type", json_integer(tag->type));
                 json_object_set_new(tag_obj, "category", json_integer(3));
+                json_object_set_new(tag_obj, "leaf", json_true());
                 json_object_set_new(tag_obj, "children", json_array());
 
                 json_array_append_new(tag_array, tag_obj);
@@ -183,6 +184,113 @@ int neu_json_encode_datalayers_get_tags_resp(void *json_object_param,
         .v.val_object = final_obj,
     } };
 
+    int ret = neu_json_encode_field(json_object_param, elems,
+                                    NEU_JSON_ELEM_SIZE(elems));
+    json_decref(final_obj);
+    return ret;
+}
+
+int neu_json_encode_datalayers_get_tag_resp(void *json_object_param,
+                                            void *param)
+{
+    neu_resp_get_sub_driver_tags_t *resp =
+        (neu_resp_get_sub_driver_tags_t *) param;
+    json_t *node_array = json_array();
+
+    for (unsigned int i = 0; i < utarray_len(resp->infos); i++) {
+        neu_resp_get_sub_driver_tags_info_t *info =
+            (neu_resp_get_sub_driver_tags_info_t *) utarray_eltptr(resp->infos,
+                                                                   i);
+
+        char node_id[37];
+        char node_uuid_input[128];
+        snprintf(node_uuid_input, sizeof(node_uuid_input), "%s", info->driver);
+        generate_fixed_uuid(node_uuid_input, node_id);
+
+        json_t *node_obj = NULL;
+        for (size_t j = 0; j < json_array_size(node_array); j++) {
+            node_obj = json_array_get(node_array, j);
+            if (strcmp(json_string_value(json_object_get(node_obj, "id")),
+                       node_id) == 0) {
+                break;
+            }
+            node_obj = NULL;
+        }
+
+        if (!node_obj) {
+            node_obj = json_object();
+            json_object_set_new(node_obj, "id", json_string(node_id));
+            json_object_set_new(node_obj, "name", json_string(info->driver));
+            json_object_set_new(node_obj, "category", json_integer(1));
+            json_t *node_children = json_array();
+            json_object_set_new(node_obj, "children", node_children);
+            json_array_append_new(node_array, node_obj);
+        }
+
+        json_t *group_obj = NULL;
+        char    group_id[37];
+        char    group_uuid_input[256];
+        snprintf(group_uuid_input, sizeof(group_uuid_input), "%s-%s",
+                 info->driver, info->group);
+        generate_fixed_uuid(group_uuid_input, group_id);
+
+        json_t *children_array = json_object_get(node_obj, "children");
+
+        for (size_t k = 0; k < json_array_size(children_array); k++) {
+            group_obj = json_array_get(children_array, k);
+            if (strcmp(json_string_value(json_object_get(group_obj, "id")),
+                       group_id) == 0) {
+                break;
+            }
+            group_obj = NULL;
+        }
+
+        if (!group_obj) {
+            group_obj = json_object();
+            json_object_set_new(group_obj, "id", json_string(group_id));
+            json_object_set_new(group_obj, "name", json_string(info->group));
+            json_object_set_new(group_obj, "category", json_integer(2));
+            json_t *group_children = json_array();
+            json_object_set_new(group_obj, "children", group_children);
+            json_array_append_new(children_array, group_obj);
+        }
+
+        json_t *tag_array = json_object_get(group_obj, "children");
+
+        if (info->tags) {
+            for (unsigned int j = 0; j < utarray_len(info->tags); j++) {
+                neu_datatag_t *tag =
+                    (neu_datatag_t *) utarray_eltptr(info->tags, j);
+
+                char tag_id[37];
+                char tag_uuid_input[512];
+                snprintf(tag_uuid_input, sizeof(tag_uuid_input), "%s-%s-%s",
+                         info->driver, info->group, tag->name);
+                generate_fixed_uuid(tag_uuid_input, tag_id);
+
+                json_t *tag_obj = json_object();
+                json_object_set_new(tag_obj, "id", json_string(tag_id));
+                json_object_set_new(tag_obj, "name", json_string(tag->name));
+                json_object_set_new(tag_obj, "type", json_integer(tag->type));
+                json_object_set_new(tag_obj, "category", json_integer(3));
+                json_object_set_new(tag_obj, "leaf", json_true());
+                json_object_set_new(tag_obj, "children", json_array());
+
+                json_array_append_new(tag_array, tag_obj);
+            }
+        }
+    }
+
+    json_t *final_obj = json_object();
+    json_object_set_new(final_obj, "error", json_integer(0));
+    json_object_set_new(final_obj, "children", node_array);
+
+    neu_json_elem_t elems[] = { {
+        .name         = NULL,
+        .t            = NEU_JSON_OBJECT,
+        .v.val_object = final_obj,
+    } };
+
     int ret = neu_json_encode_field(json_object_param, elems,
                                     NEU_JSON_ELEM_SIZE(elems));
     json_decref(final_obj);

src/parser/neu_json_datalayers.h

@@ -47,6 +47,8 @@ int neu_json_encode_datalayers_get_groups_resp(void *json_object, void *param);
 
 int neu_json_encode_datalayers_get_tags_resp(void *json_object, void *param);
 
+int neu_json_encode_datalayers_get_tag_resp(void *json_object, void *param);
+
 #ifdef __cplusplus
 }
 #endif

tests/ft/app/test_mqtt.py

@@ -62,12 +62,12 @@ TAGS = [
 ]
 FORMAT_VALUES = 0
 FORMAT_TAGS = 1
-UPLOAD_TOPIC = f"/neuron/{NODE}/upload"
-WRITE_REQ_TOPIC = f"/neuron/{NODE}/write/req"
-WRITE_RESP_TOPIC = f"/neuron/{NODE}/write/resp"
-READ_REQ_TOPIC = f"/neuron/{NODE}/read/req"
-READ_RESP_TOPIC = f"/neuron/{NODE}/read/resp"
-UPLOAD_DRV_STATE_TOPIC = f"/neuron/{NODE}/state/update"
+UPLOAD_TOPIC = f"neuron/{NODE}/upload"
+WRITE_REQ_TOPIC = f"neuron/{NODE}/write/req"
+WRITE_RESP_TOPIC = f"neuron/{NODE}/write/resp"
+READ_REQ_TOPIC = f"neuron/{NODE}/read/req"
+READ_RESP_TOPIC = f"neuron/{NODE}/read/resp"
+UPLOAD_DRV_STATE_TOPIC = f"neuron/{NODE}/state/update"
 
 
 @pytest.fixture(autouse=True, scope="class")
@@ -204,7 +204,7 @@ def conf_upload_driver_state(conf_base):
     return {
         **conf_base,
         "upload_drv_state": True,
-        "upload_drv_state_topic": f"/neuron/{NODE}/state/update",
+        "upload_drv_state_topic": f"neuron/{NODE}/state/update",
         "upload_drv_state_interval": 1,
     }
 

version

@@ -1 +1 @@
-2.12.0-alpha
+2.12.1