Merge pull request #2561 from fengzeroz/v2.12

version: 2.12.2

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.12.1
+  VERSION: 2.12.2
   UI_VERSION: 2.6.1
   BRANCH: v2.12
 

include/neuron/define.h

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

include/neuron/event/event.h

@@ -35,7 +35,7 @@ typedef struct neu_events neu_events_t;
  * thread.
  * @return the newly created event.
  */
-neu_events_t *neu_event_new(void);
+neu_events_t *neu_event_new(const char *name);
 
 /**
  * @brief Close a event.

include/neuron/msg.h

@@ -61,6 +61,8 @@ typedef enum neu_reqresp_type {
     NEU_REQ_SUBSCRIBE_GROUPS,
     NEU_REQ_GET_SUBSCRIBE_GROUP,
     NEU_RESP_GET_SUBSCRIBE_GROUP,
+    NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP,
+    NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP,
     NEU_REQ_GET_SUB_DRIVER_TAGS,
     NEU_RESP_GET_SUB_DRIVER_TAGS,
 
@@ -188,14 +190,17 @@ static const char *neu_reqresp_type_string_t[] = {
     [NEU_RESP_WRITE_TAGS]          = "NEU_RESP_WRITE_TAGS",
     [NEU_REQ_WRITE_GTAGS]          = "NEU_REQ_WRITE_GTAGS",
 
-    [NEU_REQ_SUBSCRIBE_GROUP]        = "NEU_REQ_SUBSCRIBE_GROUP",
-    [NEU_REQ_UNSUBSCRIBE_GROUP]      = "NEU_REQ_UNSUBSCRIBE_GROUP",
-    [NEU_REQ_UPDATE_SUBSCRIBE_GROUP] = "NEU_REQ_UPDATE_SUBSCRIBE_GROUP",
-    [NEU_REQ_SUBSCRIBE_GROUPS]       = "NEU_REQ_SUBSCRIBE_GROUPS",
-    [NEU_REQ_GET_SUBSCRIBE_GROUP]    = "NEU_REQ_GET_SUBSCRIBE_GROUP",
-    [NEU_RESP_GET_SUBSCRIBE_GROUP]   = "NEU_RESP_GET_SUBSCRIBE_GROUP",
-    [NEU_REQ_GET_SUB_DRIVER_TAGS]    = "NEU_REQ_GET_SUB_DRIVER_TAGS",
-    [NEU_RESP_GET_SUB_DRIVER_TAGS]   = "NEU_RESP_GET_SUB_DRIVER_TAGS",
+    [NEU_REQ_SUBSCRIBE_GROUP]            = "NEU_REQ_SUBSCRIBE_GROUP",
+    [NEU_REQ_UNSUBSCRIBE_GROUP]          = "NEU_REQ_UNSUBSCRIBE_GROUP",
+    [NEU_REQ_UPDATE_SUBSCRIBE_GROUP]     = "NEU_REQ_UPDATE_SUBSCRIBE_GROUP",
+    [NEU_REQ_SUBSCRIBE_GROUPS]           = "NEU_REQ_SUBSCRIBE_GROUPS",
+    [NEU_REQ_GET_SUBSCRIBE_GROUP]        = "NEU_REQ_GET_SUBSCRIBE_GROUP",
+    [NEU_RESP_GET_SUBSCRIBE_GROUP]       = "NEU_RESP_GET_SUBSCRIBE_GROUP",
+    [NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP] = "NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP",
+    [NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP] =
+        "NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP",
+    [NEU_REQ_GET_SUB_DRIVER_TAGS]  = "NEU_REQ_GET_SUB_DRIVER_TAGS",
+    [NEU_RESP_GET_SUB_DRIVER_TAGS] = "NEU_RESP_GET_SUB_DRIVER_TAGS",
 
     [NEU_REQ_GET_DATALAYERS_GROUPS]  = "NEU_REQ_GET_DATALAYERS_GROUPS",
     [NEU_RESP_GET_DATALAYERS_GROUPS] = "NEU_RESP_GET_DATALAYERS_GROUPS",
@@ -699,6 +704,11 @@ typedef struct neu_req_get_subscribe_group {
     char group[NEU_GROUP_NAME_LEN];
 } neu_req_get_subscribe_group_t;
 
+typedef struct neu_req_get_driver_subscribe_group {
+    char app[NEU_NODE_NAME_LEN];
+    char name[NEU_NODE_NAME_LEN];
+} neu_req_get_driver_subscribe_group_t;
+
 typedef struct {
     char app[NEU_NODE_NAME_LEN];
     char driver[NEU_NODE_NAME_LEN];
@@ -714,6 +724,13 @@ typedef struct neu_resp_subscribe_info {
     char *static_tags;
 } neu_resp_subscribe_info_t;
 
+typedef struct neu_resp_driver_subscribe_info {
+    char driver[NEU_NODE_NAME_LEN];
+    char app[NEU_NODE_NAME_LEN];
+    char group[NEU_GROUP_NAME_LEN];
+    bool subscribed;
+} neu_resp_driver_subscribe_info_t;
+
 static inline void neu_resp_subscribe_info_fini(neu_resp_subscribe_info_t *info)
 {
     free(info->params);
@@ -737,7 +754,8 @@ typedef struct {
 } neu_resp_get_sub_driver_tags_t;
 
 typedef struct neu_resp_get_subscribe_group {
-    UT_array *groups; // array neu_resp_subscribe_info_t
+    UT_array *groups; // array neu_resp_subscribe_info_t or
+                      // neu_resp_driver_subscribe_info_t
 } neu_resp_get_subscribe_group_t;
 
 typedef struct neu_req_node_setting {

plugins/modbus/modbus_rtu.c

@@ -107,7 +107,7 @@ static int driver_init(neu_plugin_t *plugin, bool load)
 {
     (void) load;
     plugin->protocol = MODBUS_PROTOCOL_RTU;
-    plugin->events   = neu_event_new();
+    plugin->events   = neu_event_new(plugin->common.name);
     plugin->stack    = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_RTU,
                                         modbus_send_msg, modbus_value_handle,
                                         modbus_write_resp);

plugins/modbus/modbus_tcp.c

@@ -110,7 +110,7 @@ static int driver_init(neu_plugin_t *plugin, bool load)
 {
     (void) load;
     plugin->protocol = MODBUS_PROTOCOL_TCP;
-    plugin->events   = neu_event_new();
+    plugin->events   = neu_event_new(plugin->common.name);
     plugin->stack    = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_TCP,
                                         modbus_send_msg, modbus_value_handle,
                                         modbus_write_resp);

plugins/monitor/monitor.c

@@ -137,7 +137,7 @@ static int start_hearbeat_timer(neu_plugin_t *plugin, uint64_t interval)
     }
 
     if (NULL == plugin->events) {
-        plugin->events = neu_event_new();
+        plugin->events = neu_event_new(plugin->common.name);
         if (NULL == plugin->events) {
             plog_error(plugin, "neu_event_new fail");
             return NEU_ERR_EINTERNAL;

plugins/mqtt/mqtt_handle.c

@@ -1491,7 +1491,9 @@ int handle_trans_data(neu_plugin_t *            plugin,
                 plugin, trans_data, plugin->config.format,
                 plugin->config.schema_vts, plugin->config.n_schema_vt,
                 static_tags, n_satic_tag, &skip_none);
-            size = strlen(json_str);
+            if (json_str != NULL) {
+                size = strlen(json_str);
+            }
         }
         if (n_satic_tag > 0) {
             mqtt_static_free(static_tags, n_satic_tag);

plugins/mqtt/mqtt_plugin_intf.c

@@ -62,7 +62,7 @@ static int start_hearbeat_timer(neu_plugin_t *plugin, uint64_t interval)
     }
 
     if (NULL == plugin->events) {
-        plugin->events = neu_event_new();
+        plugin->events = neu_event_new(plugin->common.name);
         if (NULL == plugin->events) {
             plog_error(plugin, "neu_event_new fail");
             return NEU_ERR_EINTERNAL;

plugins/restful/group_config_handle.c

@@ -406,10 +406,16 @@ void handle_grp_subscribes(nng_aio *aio)
     NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_subscribe_groups_req_t,
         neu_json_decode_subscribe_groups_req, {
-            int ret = send_subscribe_groups(aio, req);
-            if (0 != ret) {
+            int ret = 0;
+            if (req->n_group == 0) {
                 NEU_JSON_RESPONSE_ERROR(
                     ret, { neu_http_response(aio, ret, result_error); });
+            } else {
+                ret = send_subscribe_groups(aio, req);
+                if (0 != ret) {
+                    NEU_JSON_RESPONSE_ERROR(
+                        ret, { neu_http_response(aio, ret, result_error); });
+                }
             }
         })
 }
@@ -488,4 +494,111 @@ void handle_grp_get_subscribe_resp(nng_aio *                       aio,
     free(result);
     free(sub_grp_configs.groups);
     utarray_free(groups->groups);
+}
+
+void handle_grp_get_subscribes(nng_aio *aio)
+{
+    int                                  ret    = 0;
+    neu_plugin_t *                       plugin = neu_rest_get_plugin();
+    neu_req_get_driver_subscribe_group_t cmd    = { 0 };
+    neu_reqresp_head_t                   header = {
+        .ctx             = aio,
+        .type            = NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP,
+        .otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM,
+    };
+
+    NEU_VALIDATE_JWT(aio);
+
+    // required parameter
+    ret = neu_http_get_param_str(aio, "app", cmd.app, sizeof(cmd.app));
+    if (ret <= 0) {
+        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, "name", cmd.name, sizeof(cmd.name));
+    if (-1 == ret || (size_t) ret == sizeof(cmd.name)) {
+        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, {
+            neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+        });
+    }
+}
+
+void handle_grp_get_subscribes_resp(nng_aio *                       aio,
+                                    neu_resp_get_subscribe_group_t *drivers)
+{
+    char *                               result       = NULL;
+    neu_json_get_driver_subscribe_resp_t drivers_resp = { 0 };
+
+    utarray_foreach(drivers->groups, neu_resp_driver_subscribe_info_t *, group)
+    {
+        // int  index = utarray_eltidx(drivers->groups, group);
+        bool found = false;
+
+        for (int k = 0; k < drivers_resp.n_driver; k++) {
+            if (strcmp(drivers_resp.drivers[k].driver, group->driver) == 0) {
+                found = true;
+                drivers_resp.drivers[k].n_group++;
+                drivers_resp.drivers[k].groups = realloc(
+                    drivers_resp.drivers[k].groups,
+                    drivers_resp.drivers[k].n_group *
+                        sizeof(neu_json_get_driver_subscribe_resp_group_t));
+                drivers_resp.drivers[k]
+                    .groups[drivers_resp.drivers[k].n_group - 1]
+                    .group = strdup(group->group);
+                drivers_resp.drivers[k]
+                    .groups[drivers_resp.drivers[k].n_group - 1]
+                    .subscribed = group->subscribed;
+                break;
+            }
+        }
+
+        if (!found) {
+            drivers_resp.n_driver++;
+            drivers_resp.drivers = realloc(
+                drivers_resp.drivers,
+                drivers_resp.n_driver *
+                    sizeof(neu_json_get_driver_subscribe_resp_driver_t));
+
+            drivers_resp.drivers[drivers_resp.n_driver - 1].driver =
+                strdup(group->driver);
+            drivers_resp.drivers[drivers_resp.n_driver - 1].n_group = 1;
+            drivers_resp.drivers[drivers_resp.n_driver - 1].groups =
+                calloc(1, sizeof(neu_json_get_driver_subscribe_resp_group_t));
+
+            drivers_resp.drivers[drivers_resp.n_driver - 1].groups[0].group =
+                strdup(group->group);
+            drivers_resp.drivers[drivers_resp.n_driver - 1]
+                .groups[0]
+                .subscribed = group->subscribed;
+        }
+    }
+
+    neu_json_encode_by_fn(&drivers_resp,
+                          neu_json_encode_get_driver_subscribe_resp, &result);
+
+    neu_http_ok(aio, result);
+    free(result);
+
+    for (int i = 0; i < drivers_resp.n_driver; i++) {
+        free(drivers_resp.drivers[i].driver);
+        for (int j = 0; j < drivers_resp.drivers[i].n_group; j++) {
+            free(drivers_resp.drivers[i].groups[j].group);
+        }
+        free(drivers_resp.drivers[i].groups);
+    }
+    free(drivers_resp.drivers);
+
+    utarray_free(drivers->groups);
 }

plugins/restful/group_config_handle.h

@@ -39,4 +39,8 @@ void handle_grp_get_subscribe(nng_aio *aio);
 void handle_grp_get_subscribe_resp(nng_aio *                       aio,
                                    neu_resp_get_subscribe_group_t *groups);
 
+void handle_grp_get_subscribes(nng_aio *aio);
+void handle_grp_get_subscribes_resp(nng_aio *                       aio,
+                                    neu_resp_get_subscribe_group_t *groups);
+
 #endif

plugins/restful/handle.c

@@ -360,6 +360,12 @@ static struct neu_http_handler rest_handlers[] = {
         .url           = "/api/v2/subscribe",
         .value.handler = handle_grp_get_subscribe,
     },
+    {
+        .method        = NEU_HTTP_METHOD_GET,
+        .type          = NEU_HTTP_HANDLER_FUNCTION,
+        .url           = "/api/v2/subscribes",
+        .value.handler = handle_grp_get_subscribes,
+    },
     {
         .method        = NEU_HTTP_METHOD_POST,
         .type          = NEU_HTTP_HANDLER_FUNCTION,

plugins/restful/rest.c

@@ -305,6 +305,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_DRIVER_SUBSCRIBE_GROUP:
+        handle_grp_get_subscribes_resp(header->ctx,
+                                       (neu_resp_get_subscribe_group_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);

simulator/modbus_simulator.c

@@ -147,7 +147,7 @@ int main(int argc, char *argv[])
         .params.tcp_server.stop_listen  = stop_listen,
     };
 
-    events = neu_event_new();
+    events = neu_event_new("modbus_simulator");
     conn   = neu_conn_new(&param, NULL, connected, disconnected);
 
     signal(SIGINT, sig_handler);

simulator/modbus_tty_simulator.c

@@ -267,7 +267,7 @@ int main(int argc, char *argv[])
         .params.tty_client.timeout = 3000,
     };
 
-    events = neu_event_new();
+    events = neu_event_new("modbus_tty_simulator");
     conn   = neu_conn_new(&param, NULL, connected, disconnected);
 
     neu_conn_start(conn);

src/adapter/adapter.c

@@ -202,7 +202,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
     }
 
     adapter->name                    = strdup(info->name);
-    adapter->events                  = neu_event_new();
+    adapter->events                  = neu_event_new("adapter");
     adapter->state                   = NEU_NODE_RUNNING_STATE_INIT;
     adapter->handle                  = info->handle;
     adapter->cb_funs.command         = callback_funs.command;
@@ -766,6 +766,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_RESP_GET_NODE_SETTING:
     case NEU_REQ_UPDATE_GROUP:
     case NEU_RESP_GET_SUBSCRIBE_GROUP:
+    case NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP:
     case NEU_RESP_GET_DATALAYERS_GROUPS:
     case NEU_RESP_GET_DATALAYERS_TAGS:
     case NEU_RESP_GET_DATALAYERS_TAG:

src/adapter/driver/cache.c

@@ -772,7 +772,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
         }
         elem->value.type = value.type;
 
-        if (metas != NULL) {
+        if (metas != NULL && n_meta != 0) {
             if (elem->metas != NULL) {
                 free(elem->metas);
             }

src/adapter/driver/driver.c

@@ -486,6 +486,13 @@ static void update_im(neu_adapter_t *adapter, const char *group,
                 } 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);
@@ -537,13 +544,13 @@ neu_adapter_driver_t *neu_adapter_driver_create()
 {
     neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t));
 
-    driver->cache                                      = neu_driver_cache_new();
-    driver->driver_events                              = neu_event_new();
-    driver->adapter.cb_funs.driver.update              = update;
-    driver->adapter.cb_funs.driver.write_response      = write_response;
-    driver->adapter.cb_funs.driver.write_responses     = write_responses;
-    driver->adapter.cb_funs.driver.directory_response  = directory_response;
-    driver->adapter.cb_funs.driver.fup_open_response   = fup_open_response;
+    driver->cache                                     = neu_driver_cache_new();
+    driver->driver_events                             = neu_event_new("driver");
+    driver->adapter.cb_funs.driver.update             = update;
+    driver->adapter.cb_funs.driver.write_response     = write_response;
+    driver->adapter.cb_funs.driver.write_responses    = write_responses;
+    driver->adapter.cb_funs.driver.directory_response = directory_response;
+    driver->adapter.cb_funs.driver.fup_open_response  = fup_open_response;
     driver->adapter.cb_funs.driver.fdown_open_response = fdown_open_response;
     driver->adapter.cb_funs.driver.fup_data_response   = fup_data_response;
     driver->adapter.cb_funs.driver.update_im           = update_im;

src/base/msg_internal.h

@@ -30,113 +30,116 @@ extern "C" {
 
 #include "msg.h"
 
-#define NEU_REQRESP_TYPE_MAP(XX)                                       \
-    XX(NEU_RESP_ERROR, neu_resp_error_t)                               \
-    XX(NEU_REQ_READ_GROUP, neu_req_read_group_t)                       \
-    XX(NEU_RESP_READ_GROUP, neu_resp_read_group_t)                     \
-    XX(NEU_REQ_READ_GROUP_PAGINATE, neu_req_read_group_paginate_t)     \
-    XX(NEU_RESP_READ_GROUP_PAGINATE, neu_resp_read_group_paginate_t)   \
-    XX(NEU_REQ_TEST_READ_TAG, neu_req_test_read_tag_t)                 \
-    XX(NEU_RESP_TEST_READ_TAG, neu_resp_test_read_tag_t)               \
-    XX(NEU_REQ_WRITE_TAG, neu_req_write_tag_t)                         \
-    XX(NEU_REQ_WRITE_TAGS, neu_req_write_tags_t)                       \
-    XX(NEU_REQ_WRITE_GTAGS, neu_req_write_gtags_t)                     \
-    XX(NEU_REQ_SUBSCRIBE_GROUP, neu_req_subscribe_t)                   \
-    XX(NEU_REQ_UNSUBSCRIBE_GROUP, neu_req_unsubscribe_t)               \
-    XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP, neu_req_subscribe_t)            \
-    XX(NEU_REQ_SUBSCRIBE_GROUPS, neu_req_subscribe_groups_t)           \
-    XX(NEU_REQ_GET_SUBSCRIBE_GROUP, neu_req_get_subscribe_group_t)     \
-    XX(NEU_RESP_GET_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t)   \
-    XX(NEU_REQ_GET_SUB_DRIVER_TAGS, neu_req_get_sub_driver_tags_t)     \
-    XX(NEU_RESP_GET_SUB_DRIVER_TAGS, neu_resp_get_sub_driver_tags_t)   \
-    XX(NEU_REQ_NODE_INIT, neu_req_node_init_t)                         \
-    XX(NEU_REQ_NODE_UNINIT, neu_req_node_uninit_t)                     \
-    XX(NEU_RESP_NODE_UNINIT, neu_resp_node_uninit_t)                   \
-    XX(NEU_REQ_ADD_NODE, neu_req_add_node_t)                           \
-    XX(NEU_REQ_UPDATE_NODE, neu_req_update_node_t)                     \
-    XX(NEU_REQ_DEL_NODE, neu_req_del_node_t)                           \
-    XX(NEU_REQ_GET_NODE, neu_req_get_node_t)                           \
-    XX(NEU_RESP_GET_NODE, neu_resp_get_node_t)                         \
-    XX(NEU_REQ_NODE_SETTING, neu_req_node_setting_t)                   \
-    XX(NEU_REQ_GET_NODE_SETTING, neu_req_get_node_setting_t)           \
-    XX(NEU_RESP_GET_NODE_SETTING, neu_resp_get_node_setting_t)         \
-    XX(NEU_REQ_GET_NODE_STATE, neu_req_get_node_state_t)               \
-    XX(NEU_RESP_GET_NODE_STATE, neu_resp_get_node_state_t)             \
-    XX(NEU_REQ_GET_NODES_STATE, neu_req_get_nodes_state_t)             \
-    XX(NEU_RESP_GET_NODES_STATE, neu_resp_get_nodes_state_t)           \
-    XX(NEU_REQ_NODE_CTL, neu_req_node_ctl_t)                           \
-    XX(NEU_REQ_NODE_RENAME, neu_req_node_rename_t)                     \
-    XX(NEU_RESP_NODE_RENAME, neu_resp_node_rename_t)                   \
-    XX(NEU_REQ_ADD_GROUP, neu_req_add_group_t)                         \
-    XX(NEU_REQ_DEL_GROUP, neu_req_del_group_t)                         \
-    XX(NEU_REQ_UPDATE_GROUP, neu_req_update_group_t)                   \
-    XX(NEU_REQ_UPDATE_DRIVER_GROUP, neu_req_update_group_t)            \
-    XX(NEU_RESP_UPDATE_DRIVER_GROUP, neu_resp_update_group_t)          \
-    XX(NEU_REQ_GET_GROUP, neu_req_get_group_t)                         \
-    XX(NEU_RESP_GET_GROUP, neu_resp_get_group_t)                       \
-    XX(NEU_REQ_GET_DRIVER_GROUP, neu_req_get_group_t)                  \
-    XX(NEU_RESP_GET_DRIVER_GROUP, neu_resp_get_driver_group_t)         \
-    XX(NEU_REQ_ADD_TAG, neu_req_add_tag_t)                             \
-    XX(NEU_RESP_ADD_TAG, neu_resp_add_tag_t)                           \
-    XX(NEU_REQ_ADD_GTAG, neu_req_add_gtag_t)                           \
-    XX(NEU_RESP_ADD_GTAG, neu_resp_add_tag_t)                          \
-    XX(NEU_REQ_DEL_TAG, neu_req_del_tag_t)                             \
-    XX(NEU_REQ_UPDATE_TAG, neu_req_update_tag_t)                       \
-    XX(NEU_RESP_UPDATE_TAG, neu_resp_update_tag_t)                     \
-    XX(NEU_REQ_GET_TAG, neu_req_get_tag_t)                             \
-    XX(NEU_RESP_GET_TAG, neu_resp_get_tag_t)                           \
-    XX(NEU_REQ_ADD_PLUGIN, neu_req_add_plugin_t)                       \
-    XX(NEU_REQ_DEL_PLUGIN, neu_req_del_plugin_t)                       \
-    XX(NEU_REQ_UPDATE_PLUGIN, neu_req_update_plugin_t)                 \
-    XX(NEU_REQ_GET_PLUGIN, neu_req_get_plugin_t)                       \
-    XX(NEU_RESP_GET_PLUGIN, neu_resp_get_plugin_t)                     \
-    XX(NEU_REQRESP_TRANS_DATA, neu_reqresp_trans_data_t)               \
-    XX(NEU_REQRESP_NODES_STATE, neu_reqresp_nodes_state_t)             \
-    XX(NEU_REQRESP_NODE_DELETED, neu_reqresp_node_deleted_t)           \
-    XX(NEU_REQ_ADD_DRIVERS, neu_req_driver_array_t)                    \
-    XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t)           \
-    XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t)               \
-    XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t)             \
-    XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t)           \
-    XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t)                         \
-    XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t)                       \
-    XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t)                   \
-    XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t)                 \
-    XX(NEU_REQ_DRIVER_ACTION, neu_req_driver_action_t)                 \
-    XX(NEU_RESP_DRIVER_ACTION, neu_resp_driver_action_t)               \
-    XX(NEU_REQ_DRIVER_DIRECTORY, neu_req_driver_directory_t)           \
-    XX(NEU_RESP_DRIVER_DIRECTORY, neu_resp_driver_directory_t)         \
-    XX(NEU_REQ_FUP_OPEN, neu_req_fup_open_t)                           \
-    XX(NEU_RESP_FUP_OPEN, neu_resp_fup_open_t)                         \
-    XX(NEU_REQ_FUP_DATA, neu_req_fup_data_t)                           \
-    XX(NEU_RESP_FUP_DATA, neu_resp_fup_data_t)                         \
-    XX(NEU_REQ_FDOWN_OPEN, neu_req_fdown_open_t)                       \
-    XX(NEU_RESP_FDOWN_OPEN, neu_resp_fdown_open_t)                     \
-    XX(NEU_REQ_FDOWN_DATA, neu_req_fdown_data_t)                       \
-    XX(NEU_RESP_FDOWN_DATA, neu_resp_fdown_data_t)                     \
-    XX(NEU_REQ_ADD_NODE_EVENT, neu_req_add_node_t)                     \
-    XX(NEU_REQ_DEL_NODE_EVENT, neu_req_del_node_t)                     \
-    XX(NEU_REQ_NODE_CTL_EVENT, neu_req_node_ctl_t)                     \
-    XX(NEU_REQ_NODE_SETTING_EVENT, neu_req_node_setting_t)             \
-    XX(NEU_REQ_ADD_GROUP_EVENT, neu_req_add_group_t)                   \
-    XX(NEU_REQ_DEL_GROUP_EVENT, neu_req_del_group_t)                   \
-    XX(NEU_REQ_UPDATE_GROUP_EVENT, neu_req_update_group_t)             \
-    XX(NEU_REQ_ADD_TAG_EVENT, neu_req_add_tag_t)                       \
-    XX(NEU_REQ_DEL_TAG_EVENT, neu_req_del_tag_t)                       \
-    XX(NEU_REQ_UPDATE_TAG_EVENT, neu_req_update_tag_t)                 \
-    XX(NEU_REQ_ADD_GTAG_EVENT, neu_req_add_gtag_t)                     \
-    XX(NEU_REQ_ADD_PLUGIN_EVENT, neu_req_add_plugin_t)                 \
-    XX(NEU_REQ_DEL_PLUGIN_EVENT, neu_req_del_plugin_t)                 \
-    XX(NEU_REQ_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t)             \
-    XX(NEU_REQ_UNSUBSCRIBE_GROUP_EVENT, neu_req_unsubscribe_t)         \
-    XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t)      \
-    XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t)     \
-    XX(NEU_RESP_WRITE_TAGS, neu_resp_write_tags_t)                     \
-    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_REQ_GET_DATALAYERS_TAG, neu_req_get_sub_driver_tags_t)      \
+#define NEU_REQRESP_TYPE_MAP(XX)                                            \
+    XX(NEU_RESP_ERROR, neu_resp_error_t)                                    \
+    XX(NEU_REQ_READ_GROUP, neu_req_read_group_t)                            \
+    XX(NEU_RESP_READ_GROUP, neu_resp_read_group_t)                          \
+    XX(NEU_REQ_READ_GROUP_PAGINATE, neu_req_read_group_paginate_t)          \
+    XX(NEU_RESP_READ_GROUP_PAGINATE, neu_resp_read_group_paginate_t)        \
+    XX(NEU_REQ_TEST_READ_TAG, neu_req_test_read_tag_t)                      \
+    XX(NEU_RESP_TEST_READ_TAG, neu_resp_test_read_tag_t)                    \
+    XX(NEU_REQ_WRITE_TAG, neu_req_write_tag_t)                              \
+    XX(NEU_REQ_WRITE_TAGS, neu_req_write_tags_t)                            \
+    XX(NEU_REQ_WRITE_GTAGS, neu_req_write_gtags_t)                          \
+    XX(NEU_REQ_SUBSCRIBE_GROUP, neu_req_subscribe_t)                        \
+    XX(NEU_REQ_UNSUBSCRIBE_GROUP, neu_req_unsubscribe_t)                    \
+    XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP, neu_req_subscribe_t)                 \
+    XX(NEU_REQ_SUBSCRIBE_GROUPS, neu_req_subscribe_groups_t)                \
+    XX(NEU_REQ_GET_SUBSCRIBE_GROUP, neu_req_get_subscribe_group_t)          \
+    XX(NEU_RESP_GET_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t)        \
+    XX(NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP,                                  \
+       neu_req_get_driver_subscribe_group_t)                                \
+    XX(NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t) \
+    XX(NEU_REQ_GET_SUB_DRIVER_TAGS, neu_req_get_sub_driver_tags_t)          \
+    XX(NEU_RESP_GET_SUB_DRIVER_TAGS, neu_resp_get_sub_driver_tags_t)        \
+    XX(NEU_REQ_NODE_INIT, neu_req_node_init_t)                              \
+    XX(NEU_REQ_NODE_UNINIT, neu_req_node_uninit_t)                          \
+    XX(NEU_RESP_NODE_UNINIT, neu_resp_node_uninit_t)                        \
+    XX(NEU_REQ_ADD_NODE, neu_req_add_node_t)                                \
+    XX(NEU_REQ_UPDATE_NODE, neu_req_update_node_t)                          \
+    XX(NEU_REQ_DEL_NODE, neu_req_del_node_t)                                \
+    XX(NEU_REQ_GET_NODE, neu_req_get_node_t)                                \
+    XX(NEU_RESP_GET_NODE, neu_resp_get_node_t)                              \
+    XX(NEU_REQ_NODE_SETTING, neu_req_node_setting_t)                        \
+    XX(NEU_REQ_GET_NODE_SETTING, neu_req_get_node_setting_t)                \
+    XX(NEU_RESP_GET_NODE_SETTING, neu_resp_get_node_setting_t)              \
+    XX(NEU_REQ_GET_NODE_STATE, neu_req_get_node_state_t)                    \
+    XX(NEU_RESP_GET_NODE_STATE, neu_resp_get_node_state_t)                  \
+    XX(NEU_REQ_GET_NODES_STATE, neu_req_get_nodes_state_t)                  \
+    XX(NEU_RESP_GET_NODES_STATE, neu_resp_get_nodes_state_t)                \
+    XX(NEU_REQ_NODE_CTL, neu_req_node_ctl_t)                                \
+    XX(NEU_REQ_NODE_RENAME, neu_req_node_rename_t)                          \
+    XX(NEU_RESP_NODE_RENAME, neu_resp_node_rename_t)                        \
+    XX(NEU_REQ_ADD_GROUP, neu_req_add_group_t)                              \
+    XX(NEU_REQ_DEL_GROUP, neu_req_del_group_t)                              \
+    XX(NEU_REQ_UPDATE_GROUP, neu_req_update_group_t)                        \
+    XX(NEU_REQ_UPDATE_DRIVER_GROUP, neu_req_update_group_t)                 \
+    XX(NEU_RESP_UPDATE_DRIVER_GROUP, neu_resp_update_group_t)               \
+    XX(NEU_REQ_GET_GROUP, neu_req_get_group_t)                              \
+    XX(NEU_RESP_GET_GROUP, neu_resp_get_group_t)                            \
+    XX(NEU_REQ_GET_DRIVER_GROUP, neu_req_get_group_t)                       \
+    XX(NEU_RESP_GET_DRIVER_GROUP, neu_resp_get_driver_group_t)              \
+    XX(NEU_REQ_ADD_TAG, neu_req_add_tag_t)                                  \
+    XX(NEU_RESP_ADD_TAG, neu_resp_add_tag_t)                                \
+    XX(NEU_REQ_ADD_GTAG, neu_req_add_gtag_t)                                \
+    XX(NEU_RESP_ADD_GTAG, neu_resp_add_tag_t)                               \
+    XX(NEU_REQ_DEL_TAG, neu_req_del_tag_t)                                  \
+    XX(NEU_REQ_UPDATE_TAG, neu_req_update_tag_t)                            \
+    XX(NEU_RESP_UPDATE_TAG, neu_resp_update_tag_t)                          \
+    XX(NEU_REQ_GET_TAG, neu_req_get_tag_t)                                  \
+    XX(NEU_RESP_GET_TAG, neu_resp_get_tag_t)                                \
+    XX(NEU_REQ_ADD_PLUGIN, neu_req_add_plugin_t)                            \
+    XX(NEU_REQ_DEL_PLUGIN, neu_req_del_plugin_t)                            \
+    XX(NEU_REQ_UPDATE_PLUGIN, neu_req_update_plugin_t)                      \
+    XX(NEU_REQ_GET_PLUGIN, neu_req_get_plugin_t)                            \
+    XX(NEU_RESP_GET_PLUGIN, neu_resp_get_plugin_t)                          \
+    XX(NEU_REQRESP_TRANS_DATA, neu_reqresp_trans_data_t)                    \
+    XX(NEU_REQRESP_NODES_STATE, neu_reqresp_nodes_state_t)                  \
+    XX(NEU_REQRESP_NODE_DELETED, neu_reqresp_node_deleted_t)                \
+    XX(NEU_REQ_ADD_DRIVERS, neu_req_driver_array_t)                         \
+    XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t)                \
+    XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t)                    \
+    XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t)                  \
+    XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t)                \
+    XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t)                              \
+    XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t)                            \
+    XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t)                        \
+    XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t)                      \
+    XX(NEU_REQ_DRIVER_ACTION, neu_req_driver_action_t)                      \
+    XX(NEU_RESP_DRIVER_ACTION, neu_resp_driver_action_t)                    \
+    XX(NEU_REQ_DRIVER_DIRECTORY, neu_req_driver_directory_t)                \
+    XX(NEU_RESP_DRIVER_DIRECTORY, neu_resp_driver_directory_t)              \
+    XX(NEU_REQ_FUP_OPEN, neu_req_fup_open_t)                                \
+    XX(NEU_RESP_FUP_OPEN, neu_resp_fup_open_t)                              \
+    XX(NEU_REQ_FUP_DATA, neu_req_fup_data_t)                                \
+    XX(NEU_RESP_FUP_DATA, neu_resp_fup_data_t)                              \
+    XX(NEU_REQ_FDOWN_OPEN, neu_req_fdown_open_t)                            \
+    XX(NEU_RESP_FDOWN_OPEN, neu_resp_fdown_open_t)                          \
+    XX(NEU_REQ_FDOWN_DATA, neu_req_fdown_data_t)                            \
+    XX(NEU_RESP_FDOWN_DATA, neu_resp_fdown_data_t)                          \
+    XX(NEU_REQ_ADD_NODE_EVENT, neu_req_add_node_t)                          \
+    XX(NEU_REQ_DEL_NODE_EVENT, neu_req_del_node_t)                          \
+    XX(NEU_REQ_NODE_CTL_EVENT, neu_req_node_ctl_t)                          \
+    XX(NEU_REQ_NODE_SETTING_EVENT, neu_req_node_setting_t)                  \
+    XX(NEU_REQ_ADD_GROUP_EVENT, neu_req_add_group_t)                        \
+    XX(NEU_REQ_DEL_GROUP_EVENT, neu_req_del_group_t)                        \
+    XX(NEU_REQ_UPDATE_GROUP_EVENT, neu_req_update_group_t)                  \
+    XX(NEU_REQ_ADD_TAG_EVENT, neu_req_add_tag_t)                            \
+    XX(NEU_REQ_DEL_TAG_EVENT, neu_req_del_tag_t)                            \
+    XX(NEU_REQ_UPDATE_TAG_EVENT, neu_req_update_tag_t)                      \
+    XX(NEU_REQ_ADD_GTAG_EVENT, neu_req_add_gtag_t)                          \
+    XX(NEU_REQ_ADD_PLUGIN_EVENT, neu_req_add_plugin_t)                      \
+    XX(NEU_REQ_DEL_PLUGIN_EVENT, neu_req_del_plugin_t)                      \
+    XX(NEU_REQ_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t)                  \
+    XX(NEU_REQ_UNSUBSCRIBE_GROUP_EVENT, neu_req_unsubscribe_t)              \
+    XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t)           \
+    XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t)          \
+    XX(NEU_RESP_WRITE_TAGS, neu_resp_write_tags_t)                          \
+    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_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/connection/connection.c

@@ -503,7 +503,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                 conn->param.log,
                 "tcp conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
                 conn->fd, len, ret, strerror(errno), errno);
-            if (ret == 0 || (ret == -1 && errno != EAGAIN)) {
+            if (ret == 0 || (conn->connection_ok && ret == -1 && errno != EAGAIN)) {
                 conn_disconnect(conn);
             }
         }

src/connection/connection_eth.c

@@ -137,7 +137,7 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
                 conn_eth->ic = &in_conns[i];
 
                 in_conns[i].interface = strdup(interface);
-                in_conns[i].events    = neu_event_new();
+                in_conns[i].events    = neu_event_new("eth_conn");
                 in_conns[i].count += 1;
 
                 get_mac(conn_eth, interface);

src/connection/mqtt_client.c

@@ -907,7 +907,7 @@ static inline int client_start_timer(neu_mqtt_client_t *client)
         return 0;
     }
 
-    events = neu_event_new();
+    events = neu_event_new("mqtt_client");
     if (NULL == events) {
         return -1;
     }

src/core/manager.c

@@ -98,7 +98,7 @@ neu_manager_t *neu_manager_create()
         .type        = NEU_EVENT_TIMER_NOBLOCK,
     };
 
-    manager->events            = neu_event_new();
+    manager->events            = neu_event_new("manager");
     manager->plugin_manager    = neu_plugin_manager_create();
     manager->node_manager      = neu_node_manager_create();
     manager->subscribe_manager = neu_subscribe_manager_create();
@@ -1010,6 +1010,18 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         reply(manager, header, &resp);
         break;
     }
+    case NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP: {
+        neu_req_get_driver_subscribe_group_t *cmd =
+            (neu_req_get_driver_subscribe_group_t *) &header[1];
+        UT_array *groups =
+            neu_manager_get_driver_groups(manager, cmd->app, cmd->name);
+        neu_resp_get_subscribe_group_t resp = { .groups = groups };
+
+        strcpy(header->receiver, header->sender);
+        header->type = NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP;
+        reply(manager, header, &resp);
+        break;
+    }
     case NEU_REQ_GET_SUB_DRIVER_TAGS: {
         neu_req_get_sub_driver_tags_t *cmd =
             (neu_req_get_sub_driver_tags_t *) &header[1];

src/core/manager_internal.c

@@ -380,6 +380,55 @@ UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
     return subs;
 }
 
+UT_array *neu_manager_get_driver_groups(neu_manager_t *manager, const char *app,
+                                        const char *name)
+{
+    UT_icd icd = { sizeof(neu_resp_driver_subscribe_info_t), NULL, NULL, NULL };
+    UT_array *result = NULL;
+
+    UT_array *sub_groups =
+        neu_subscribe_manager_get(manager->subscribe_manager, app, NULL, NULL);
+    UT_array *driver_groups = neu_manager_get_driver_group(manager);
+
+    utarray_new(result, &icd);
+    utarray_foreach(driver_groups, neu_resp_driver_group_info_t *, grp)
+    {
+        neu_resp_driver_subscribe_info_t info = { 0 };
+
+        if (name != NULL && strlen(name) > 0) {
+            if (strstr(grp->driver, name) != NULL ||
+                strstr(grp->group, name) != NULL) {
+                strcpy(info.app, app);
+                strcpy(info.driver, grp->driver);
+                strcpy(info.group, grp->group);
+                info.subscribed = false;
+            }
+        } else {
+            strcpy(info.app, app);
+            strcpy(info.driver, grp->driver);
+            strcpy(info.group, grp->group);
+            info.subscribed = false;
+        }
+
+        if (strlen(info.driver) > 0 && strlen(info.group) > 0) {
+            utarray_foreach(sub_groups, neu_resp_subscribe_info_t *, sub)
+            {
+                if (strcmp(sub->driver, info.driver) == 0 &&
+                    strcmp(sub->group, info.group) == 0) {
+                    info.subscribed = true;
+                    break;
+                }
+            }
+
+            utarray_push_back(result, &info);
+        }
+    }
+
+    utarray_free(sub_groups);
+    utarray_free(driver_groups);
+    return result;
+}
+
 int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
                               neu_persist_node_info_t *info)
 {

src/core/manager_internal.h

@@ -85,6 +85,8 @@ UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
                                               const char *   app,
                                               const char *   driver,
                                               const char *   group);
+UT_array *neu_manager_get_driver_groups(neu_manager_t *manager, const char *app,
+                                        const char *name);
 
 int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
                               neu_persist_node_info_t *info);

src/event/event_linux.c

@@ -72,6 +72,7 @@ struct neu_events {
     int       epoll_fd;
     pthread_t thread;
     bool      stop;
+    char *    name;
 
     pthread_mutex_t   mtx;
     int               n_event;
@@ -122,8 +123,8 @@ static void *event_loop(void *arg)
         }
 
         if (ret == -1 || events->stop) {
-            zlog_warn(neuron, "event loop exit, errno: %s(%d), stop: %d",
-                      strerror(errno), errno, events->stop);
+            zlog_warn(neuron, "event loop(%s) exit, errno: %s(%d), stop: %d",
+                      events->name, strerror(errno), errno, events->stop);
             break;
         }
 
@@ -177,17 +178,18 @@ static void *event_loop(void *arg)
     return NULL;
 };
 
-neu_events_t *neu_event_new(void)
+neu_events_t *neu_event_new(const char *name)
 {
     neu_events_t *events = calloc(1, sizeof(struct neu_events));
 
     events->epoll_fd = epoll_create(1);
 
-    nlog_notice("create epoll: %d(%d)", events->epoll_fd, errno);
+    nlog_notice("create epoll(%s): %d(%d)", name, events->epoll_fd, errno);
     assert(events->epoll_fd > 0);
 
     events->stop    = false;
     events->n_event = 0;
+    events->name    = strdup(name);
     pthread_mutex_init(&events->mtx, NULL);
 
     pthread_create(&events->thread, NULL, event_loop, events);
@@ -199,6 +201,7 @@ int neu_event_close(neu_events_t *events)
 {
     events->stop = true;
     close(events->epoll_fd);
+    free(events->name);
 
     pthread_join(events->thread, NULL);
     pthread_mutex_destroy(&events->mtx);
@@ -220,7 +223,8 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
     };
     int index = get_free_event(events);
     if (index < 0) {
-        zlog_fatal(neuron, "no free event: %d", events->epoll_fd);
+        zlog_fatal(neuron, "no free event(%s): %d", events->name,
+                   events->epoll_fd);
     }
     assert(index >= 0);
 
@@ -251,18 +255,19 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
 
     zlog_notice(neuron,
                 "add timer, second: %" PRId64 ", millisecond: %" PRId64
-                ", timer: %d in epoll %d, "
+                ", timer: %d in epoll(%s) %d, "
                 "ret: %d, index: %d",
-                timer.second, timer.millisecond, timer_fd, events->epoll_fd,
-                ret, index);
+                timer.second, timer.millisecond, timer_fd, events->name,
+                events->epoll_fd, ret, index);
 
     return timer_ctx;
 }
 
 int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
 {
-    zlog_notice(neuron, "del timer: %d from epoll: %d, index: %d", timer->fd,
-                events->epoll_fd, timer->event_data->index);
+    zlog_notice(neuron, "del timer: %d from epoll(%s): %d, index: %d",
+                timer->fd, events->name, events->epoll_fd,
+                timer->event_data->index);
 
     timer->stop = true;
     epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL);
@@ -281,8 +286,8 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
     int ret   = 0;
     int index = get_free_event(events);
 
-    nlog_notice("add io, fd: %d, epoll: %d, index: %d", io.fd, events->epoll_fd,
-                index);
+    nlog_notice("add io, fd: %d, epoll(%s): %d, index: %d", io.fd, events->name,
+                events->epoll_fd, index);
     assert(index >= 0);
 
     neu_event_io_t *io_ctx   = &events->event_datas[index].ctx.io;
@@ -303,8 +308,8 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
 
     ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, io.fd, &event);
 
-    nlog_notice("add io, fd: %d, epoll: %d, ret: %d(%d), index: %d", io.fd,
-                events->epoll_fd, ret, errno, index);
+    nlog_notice("add io, fd: %d, epoll(%s): %d, ret: %d(%d), index: %d", io.fd,
+                events->name, events->epoll_fd, ret, errno, index);
     assert(ret == 0);
 
     return io_ctx;
@@ -316,8 +321,8 @@ int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
         return 0;
     }
 
-    zlog_notice(neuron, "del io: %d from epoll: %d, index: %d", io->fd,
-                events->epoll_fd, io->event_data->index);
+    zlog_notice(neuron, "del io: %d from epoll(%s): %d, index: %d", io->fd,
+                events->name, events->epoll_fd, io->event_data->index);
 
     epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, io->fd, NULL);
     free_event(events, io->event_data->index);

src/otel/otel_manager.c

@@ -1113,7 +1113,7 @@ static int otel_timer_cb(void *data)
 void neu_otel_start()
 {
     if (otel_event == NULL) {
-        otel_event = neu_event_new();
+        otel_event = neu_event_new("otel_event");
     }
 
     if (otel_timer == NULL) {

src/parser/neu_json_group_config.c

@@ -405,6 +405,50 @@ int neu_json_encode_get_subscribe_resp(void *object, void *param)
     return ret;
 }
 
+int neu_json_encode_get_driver_subscribe_resp(void *object, void *param)
+{
+
+    neu_json_get_driver_subscribe_resp_t *resp =
+        (neu_json_get_driver_subscribe_resp_t *) param;
+
+    void *driver_array = neu_json_array();
+    for (int i = 0; i < resp->n_driver; i++) {
+        if (resp->drivers[i].n_group == 0) {
+            continue;
+        }
+
+        json_t *ob          = json_object();
+        void *  group_array = neu_json_array();
+
+        json_object_set_new(ob, "name", json_string(resp->drivers[i].driver));
+
+        for (int j = 0; j < resp->drivers[i].n_group; j++) {
+            json_t *group_ob = json_object();
+
+            json_object_set_new(group_ob, "name",
+                                json_string(resp->drivers[i].groups[j].group));
+            json_object_set_new(
+                group_ob, "subscribed",
+                json_boolean(resp->drivers[i].groups[j].subscribed));
+
+            json_array_append_new(group_array, group_ob);
+        }
+
+        json_object_set_new(ob, "groups", group_array);
+
+        json_array_append_new(driver_array, ob);
+    }
+
+    neu_json_elem_t resp_elems[] = { {
+        .name         = "drivers",
+        .t            = NEU_JSON_OBJECT,
+        .v.val_object = driver_array,
+    } };
+
+    return neu_json_encode_field(object, resp_elems,
+                                 NEU_JSON_ELEM_SIZE(resp_elems));
+}
+
 static inline int dump_params(void *root, char **const result)
 {
     return neu_json_dump_key(root, "params", result, false);

src/parser/neu_json_group_config.h

@@ -113,6 +113,24 @@ typedef struct {
 
 int neu_json_encode_get_subscribe_resp(void *json_object, void *param);
 
+typedef struct {
+    char *group;
+    bool  subscribed;
+} neu_json_get_driver_subscribe_resp_group_t;
+
+typedef struct {
+    char *                                      driver;
+    int                                         n_group;
+    neu_json_get_driver_subscribe_resp_group_t *groups;
+} neu_json_get_driver_subscribe_resp_driver_t;
+
+typedef struct {
+    int                                          n_driver;
+    neu_json_get_driver_subscribe_resp_driver_t *drivers;
+} neu_json_get_driver_subscribe_resp_t;
+
+int neu_json_encode_get_driver_subscribe_resp(void *json_object, void *param);
+
 typedef struct {
     char *group;
     char *app;

src/utils/json.c

@@ -534,6 +534,10 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
                     ele->t = NEU_JSON_ARRAY_BOOL;
                     break;
                 }
+                if (json_is_object(value)) {
+                    ele->t = NEU_JSON_OBJECT;
+                    break;
+                }
             }
         } else if (json_is_object(ob)) {
             ele->t = NEU_JSON_OBJECT;

version

@@ -1 +1 @@
-2.12.1
+2.12.2