Add message NEU_REQRESP_GET_DATATAGS for get datatag table

* Add two message NEU_REQRESP_GET_DATATAGS and NEU_REQRESP_DATATAGS
* Add get datatag table function in bus manager and datatag manager

include/neu_adapter.h

@@ -24,6 +24,7 @@
 #include <string.h>
 #include <unistd.h>
 
+#include "neu_datatag_table.h"
 #include "neu_tag_group_config.h"
 #include "neu_types.h"
 
@@ -44,6 +45,8 @@ typedef enum neu_reqresp_type {
     NEU_REQRESP_NODES,
     NEU_REQRESP_GET_GRP_CONFIGS,
     NEU_REQRESP_GRP_CONFIGS,
+    NEU_REQRESP_GET_DATATAGS,
+    NEU_REQRESP_DATATAGS,
 } neu_reqresp_type_e;
 
 typedef uint32_t neu_node_id_t;
@@ -94,7 +97,7 @@ typedef struct neu_reqresp_nodes {
 
 /* NEU_REQRESP_GET_GRP_CONFIGS */
 typedef struct neu_cmd_get_grp_configs {
-    neu_node_id_t node_id; // vector of neu_node_info_t
+    neu_node_id_t node_id; // get group configs of this node
 } neu_cmd_get_grp_configs_t;
 
 /* NEU_REQRESP_GRP_CONFIGS */
@@ -102,6 +105,16 @@ typedef struct neu_reqresp_grp_configs {
     vector_t grp_configs; // vector of neu_taggrp_config_t pointer
 } neu_reqresp_grp_configs_t;
 
+/* NEU_REQRESP_GET_DATATAGS */
+typedef struct neu_cmd_get_datatags {
+    neu_node_id_t node_id; // get datatags of this node
+} neu_cmd_get_datatags_t;
+
+/* NEU_REQRESP_DATATAGS */
+typedef struct neu_reqresp_datatags {
+    neu_datatag_table_t *datatag_tbl; // datatag table of a node
+} neu_reqresp_datatags_t;
+
 typedef struct neu_request {
     uint32_t           req_id;
     neu_reqresp_type_e req_type;

include/neu_datatag_table.h

@@ -21,7 +21,6 @@
 #define __DATATAG_TABLE_H__
 
 #include "neu_types.h"
-#include "utils/idhash.h"
 
 typedef uint32_t                 datatag_id_t;
 typedef struct neu_datatag_table neu_datatag_table_t;

src/adapter/adapter.c

@@ -303,10 +303,12 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
             msg_inplace_data_init(msg_ptr, MSG_CMD_WRITE_DATA,
                                   sizeof(write_data_cmd_t));
 
-            cmd_ptr             = (write_data_cmd_t *) msg_get_buf_ptr(msg_ptr);
-            cmd_ptr->grp_config = write_cmd->grp_config;
-            cmd_ptr->addr       = write_cmd->addr;
-            cmd_ptr->databuf    = databuf;
+            cmd_ptr            = (write_data_cmd_t *) msg_get_buf_ptr(msg_ptr);
+            cmd_ptr->sender_id = adapter->id;
+            cmd_ptr->dst_node_id = write_cmd->dst_node_id;
+            cmd_ptr->grp_config  = write_cmd->grp_config;
+            cmd_ptr->addr        = write_cmd->addr;
+            cmd_ptr->databuf     = databuf;
             nng_sendmsg(adapter->sock, write_msg, 0);
         }
         break;
@@ -382,6 +384,42 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
         break;
     }
 
+    case NEU_REQRESP_GET_DATATAGS: {
+        neu_response_t *        result;
+        neu_reqresp_datatags_t *resp_datatags;
+
+        assert(cmd->buf_len == sizeof(neu_cmd_get_datatags_t));
+        resp_datatags = malloc(sizeof(neu_reqresp_datatags_t));
+        if (resp_datatags == NULL) {
+            log_error("Failed to allocate result of datatags");
+            rv = -1;
+            break;
+        }
+        result = malloc(sizeof(neu_response_t));
+        if (result == NULL) {
+            log_error("Failed to allocate result for get datatags");
+            free(resp_datatags);
+            rv = -1;
+            break;
+        }
+
+        neu_cmd_get_datatags_t *datatags_cmd;
+        neu_datatag_table_t *   tag_table;
+        datatags_cmd = (neu_cmd_get_datatags_t *) cmd->buf;
+        tag_table    = neu_manager_get_datatag_tbl(adapter->manager,
+                                                datatags_cmd->node_id);
+        resp_datatags->datatag_tbl = tag_table;
+
+        result->resp_type = NEU_REQRESP_DATATAGS;
+        result->req_id    = cmd->req_id;
+        result->buf_len   = sizeof(neu_reqresp_datatags_t);
+        result->buf       = resp_datatags;
+        if (p_result != NULL) {
+            *p_result = result;
+        }
+        break;
+    }
+
     default:
         rv = -1;
         break;

src/core/neu_datatag_manager.c

@@ -108,6 +108,14 @@ neu_datatag_manager_t *neu_datatag_mng_create(neu_adapter_t *adapter)
         goto init_p_configs_fail;
     }
 
+    neu_datatag_table_t *tag_table;
+    tag_table = neu_datatag_tbl_create();
+    if (tag_table == NULL) {
+        log_error("Failed to create datatag table in datatag manager");
+        goto create_tag_table_fail;
+    }
+    datatag_manager->tag_table = tag_table;
+
     datatag_manager->adapter = adapter;
     if (nng_mtx_alloc(&datatag_manager->mtx) != 0) {
         log_error("Failed to initialize mutex in datatag manager");
@@ -116,6 +124,8 @@ neu_datatag_manager_t *neu_datatag_mng_create(neu_adapter_t *adapter)
     return datatag_manager;
 
 alloc_mtx_fail:
+    neu_datatag_tbl_destroy(tag_table);
+create_tag_table_fail:
     vector_uninit(&datatag_manager->p_configs);
 init_p_configs_fail:
     free(datatag_manager);
@@ -129,6 +139,7 @@ void neu_datatag_mng_destroy(neu_datatag_manager_t *datatag_manager)
     }
 
     nng_mtx_free(datatag_manager->mtx);
+    neu_datatag_tbl_destroy(datatag_manager->tag_table);
     free_all_taggrp_config(&datatag_manager->p_configs);
     vector_uninit(&datatag_manager->p_configs);
     free(datatag_manager);
@@ -279,6 +290,19 @@ int neu_datatag_mng_ref_all_grp_configs(neu_datatag_manager_t *datatag_manager,
     return rv;
 }
 
+neu_datatag_table_t *
+neu_datatag_mng_get_datatag_tbl(neu_datatag_manager_t *datatag_manager)
+{
+    if (datatag_manager == NULL) {
+        return NULL;
+    }
+
+    return datatag_manager->tag_table;
+}
+
+/*
+ * Functions for datatag group config
+ */
 neu_taggrp_config_t *neu_taggrp_cfg_new(char *config_name)
 {
     int                  rv;

src/core/neu_datatag_manager.h

@@ -41,5 +41,7 @@ neu_taggrp_config_t *
                                    const char *           config_name);
 int neu_datatag_mng_ref_all_grp_configs(neu_datatag_manager_t *datatag_manager,
                                         vector_t *             grp_configs);
+neu_datatag_table_t *
+neu_datatag_mng_get_datatag_tbl(neu_datatag_manager_t *datatag_manager);
 
 #endif

src/core/neu_datatag_table.c

@@ -18,6 +18,7 @@
  **/
 
 #include "neu_datatag_table.h"
+#include "utils/idhash.h"
 #include "utils/mem_alloc.h"
 #include <nng/nng.h>
 #include <nng/supplemental/util/platform.h>

src/core/neu_manager.c

@@ -406,10 +406,12 @@ static adapter_id_t adapter_id_from_node_id(neu_node_id_t node_id)
     return node_id;
 }
 
+/*
 static neu_node_id_t adapter_id_to_node_id(adapter_id_t adapter_id)
 {
     return adapter_id;
 }
+*/
 
 static void manager_bind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg)
 {
@@ -1040,3 +1042,29 @@ int neu_manager_get_grp_configs(neu_manager_t *manager, neu_node_id_t node_id,
 
     return rv;
 }
+
+neu_datatag_table_t *neu_manager_get_datatag_tbl(neu_manager_t *manager,
+                                                 neu_node_id_t  node_id)
+{
+    adapter_reg_entity_t *reg_entity;
+    neu_datatag_table_t * tag_table;
+    adapter_id_t          adapter_id;
+
+    if (manager == NULL) {
+        log_error("get datatag table with NULL manager");
+        return NULL;
+    }
+
+    tag_table  = NULL;
+    adapter_id = adapter_id_from_node_id(node_id);
+    VECTOR_FOR_EACH(&manager->reg_adapters, iter)
+    {
+        reg_entity = (adapter_reg_entity_t *) iterator_get(&iter);
+        if (reg_entity->adapter_id == adapter_id) {
+            tag_table =
+                neu_datatag_mng_get_datatag_tbl(reg_entity->datatag_manager);
+        }
+    }
+
+    return tag_table;
+}

src/core/neu_manager.h

@@ -35,4 +35,6 @@ int neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e node_type,
                           vector_t *result_nodes);
 int neu_manager_get_grp_configs(neu_manager_t *manager, neu_node_id_t node_id,
                                 vector_t *result_grp_configs);
+neu_datatag_table_t *neu_manager_get_datatag_tbl(neu_manager_t *manager,
+                                                 neu_node_id_t  node_id);
 #endif