✨ feat(integrate utable): deps/utable

deps/utable/CMakeLists.txt

@@ -0,0 +1,17 @@
+set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -lm")
+
+add_library(utable SHARED utable.c  utable_ipfix_exporter.c)
+target_link_libraries(utable base64 nmx_pool cjson-static yyjson mpack)
+
+add_executable(ipfix_exporter_example
+    utable_ipfix_exporter.c
+    ipfix_exporter_example.cpp
+)
+
+target_link_libraries(
+  ipfix_exporter_example
+  utable
+  pthread
+)
+
+add_subdirectory(test)

deps/utable/ipfix_exporter_example.cpp

@@ -0,0 +1,197 @@
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <arpa/inet.h>
+#include <unistd.h>
+#include <pthread.h>
+
+#include "utable.h"
+#include "cjson/cJSON.h"
+
+#define THREAD_MAX 8
+#define TEMPLATE_MAX 13
+struct ipfix_template_id_list
+{
+    const char *template_name;
+    int template_id;
+};
+
+struct ipfix_template_id_list template_id_list[TEMPLATE_MAX] = {
+    {"BASE", 0},
+    {"SSL", 0},
+    {"HTTP", 0},
+    {"MAIL", 0},
+    {"DNS", 0},
+    {"DTLS", 0},
+    {"QUIC", 0},
+    {"FTP", 0},
+    {"SIP", 0},
+    {"RTP", 0},
+    {"SSH", 0},
+    {"RDP", 0},
+    {"Stratum", 0}};
+int g_udp_sock_fd = 0;
+const char *ipfix_schema_json_path = NULL;
+struct ipfix_exporter_schema *g_ipfix_schema = NULL;
+
+static int ipfix_exporter_get_socket_fd(char *collector_ip, uint16_t collector_port)
+{
+    int sock_fd = socket(AF_INET, SOCK_DGRAM, 0);
+    if (sock_fd <= 0)
+    {
+        return -1;
+    }
+
+    struct sockaddr_in addr;
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(collector_port);
+    addr.sin_addr.s_addr = inet_addr(collector_ip);
+
+    if (connect(sock_fd, (struct sockaddr *)&addr, sizeof(addr)) == -1)
+    {
+        printf("connect error, illegal collector ip or port\n");
+        printf("expample: ./ipfix_exporter_example 127.0.0.1 4397");
+        close(sock_fd);
+        return -1;
+    }
+
+    return sock_fd;
+}
+
+void *ipfix_template_send_thread_loop(void *arg)
+{
+    int interval_s = (*(int *)arg);
+    while (1)
+    {
+        size_t blob_len = 0;
+        char *blob = NULL;
+        for (int i = 0; i < THREAD_MAX; i++)
+        {
+            blob = (char *)utable_ipfix_template_flow_get0(g_ipfix_schema, i, &blob_len);
+            send(g_udp_sock_fd, blob, blob_len, 0);
+            blob = NULL;
+        }
+
+        sleep(interval_s);
+    }
+
+    return NULL;
+}
+
+extern "C" int load_file_to_memory(const char *file_name, unsigned char **pp_out, size_t *out_sz);
+void ipfix_exporter_test_utable_init(struct utable *table, int index, const char *file_name)
+{
+    size_t json_size = 0;
+    unsigned char *json_str = NULL;
+    load_file_to_memory(file_name, &json_str, &json_size);
+    if (json_str == NULL || json_size == 0)
+    {
+        return;
+    }
+
+    cJSON *root = NULL;
+    root = cJSON_Parse((const char *)json_str);
+
+    cJSON *template_item = cJSON_GetArrayItem(cJSON_GetObjectItem(root, "templates"), index);
+    cJSON *template_key_array = cJSON_GetObjectItem(template_item, "elements");
+    for (int i = 0; i < cJSON_GetArraySize(template_key_array); i++)
+    {
+        char *template_key = cJSON_GetArrayItem(template_key_array, i)->valuestring;
+        cJSON *elements_array = cJSON_GetObjectItem(root, template_key);
+        for (int j = 0; j < cJSON_GetArraySize(elements_array); j++)
+        {
+            cJSON *element = cJSON_GetArrayItem(elements_array, j);
+            char *element_key = cJSON_GetObjectItem(element, "element_name")->valuestring;
+            if (strcmp(cJSON_GetObjectItem(element, "element_type")->valuestring, "string") == 0)
+            {
+                char temp[128] = {0};
+                snprintf(temp, 128, "%s_%s_%d", element_key, "string", cJSON_GetObjectItem(element, "element_id")->valueint);
+                utable_add_cstring(table, element_key, temp, strlen(temp));
+            }
+            else if (strcmp(cJSON_GetObjectItem(element, "element_type")->valuestring, "unsigned64") == 0 ||
+                     strcmp(cJSON_GetObjectItem(element, "element_type")->valuestring, "unsigned32") == 0 ||
+                     strcmp(cJSON_GetObjectItem(element, "element_type")->valuestring, "unsigned16") == 0 ||
+                     strcmp(cJSON_GetObjectItem(element, "element_type")->valuestring, "unsigned8") == 0)
+            {
+                utable_add_integer(table, element_key, cJSON_GetObjectItem(element, "element_id")->valueint);
+            }
+        }
+    }
+
+    free(json_str);
+    cJSON_Delete(root);
+}
+
+void *ipfix_worker_thread_data_flow_send(void *arg)
+{
+    uint16_t worker_id = (*(uint16_t *)arg);
+    while (1)
+    {
+        for (int i = 0; i < TEMPLATE_MAX; i++)
+        {
+            struct utable *table = utable_new();
+            ipfix_exporter_test_utable_init(table, i, ipfix_schema_json_path);
+            utable_delete_item(table, "decoded_as");
+            utable_add_cstring(table, "decoded_as", template_id_list[i].template_name, strlen(template_id_list[i].template_name));
+
+            size_t blob_len = 0;
+            char *blob = NULL;
+            utable_ipfix_data_flow_exporter(table, g_ipfix_schema, template_id_list[i].template_id, worker_id, &blob, &blob_len);
+            send(g_udp_sock_fd, blob, blob_len, 0);
+            free(blob);
+            blob = NULL;
+            utable_free(table);
+        }
+
+        sleep(5);
+    }
+
+    return NULL;
+}
+
+// ./ipfix_exporter_example ipfix_schema.json 127.0.0.1 4397
+extern "C" int main(int argc, char *argv[])
+{
+    if (argc != 4)
+    {
+        printf("expample: ./ipfix_exporter_example ipfix_schema.json 127.0.0.1 4397\n");
+        return -1;
+    }
+
+    ipfix_schema_json_path = argv[1];
+    g_ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, 1, THREAD_MAX);
+    if (g_ipfix_schema == NULL)
+    {
+        printf("ipfix_exporter_schema_init error, illegal ipfix_schema_json_path: %s\n", ipfix_schema_json_path);
+        return -1;
+    }
+
+    for (int i = 0; i < TEMPLATE_MAX; i++)
+    {
+        template_id_list[i].template_id = utable_ipfix_template_get(g_ipfix_schema, template_id_list[i].template_name);
+    }
+
+    g_udp_sock_fd = ipfix_exporter_get_socket_fd(argv[2], atoi(argv[3]));
+    int interval_s = 100;
+    pthread_t template_thread_id;
+    pthread_create(&template_thread_id, NULL, ipfix_template_send_thread_loop, (void *)&interval_s);
+
+    uint16_t worker_id[THREAD_MAX];
+    pthread_t pid[THREAD_MAX];
+    for (int i = 0; i < THREAD_MAX; i++)
+    {
+        worker_id[i] = i;
+        pthread_create(&pid[i], NULL, ipfix_worker_thread_data_flow_send, (void *)&worker_id[i]);
+    }
+
+    sleep(1000);
+    utable_ipfix_exporter_schema_free(g_ipfix_schema);
+    pthread_join(template_thread_id, NULL);
+    for (int i = 0; i < THREAD_MAX; i++)
+    {
+        pthread_join(pid[i], NULL);
+    }
+    close(g_udp_sock_fd);
+    return 0;
+}

deps/utable/test/CMakeLists.txt

@@ -0,0 +1,31 @@
+include_directories(${CMAKE_SOURCE_DIR}/deps)
+
+add_executable(gtest_utable
+    unit_test_utable.cpp
+)
+
+target_link_libraries(
+  gtest_utable
+  utable
+  gtest
+)
+
+add_executable(gtest_ipfix_exporter
+    unit_test_ipfix_exporter.cpp
+)
+
+target_link_libraries(
+  gtest_ipfix_exporter
+  utable
+  gtest
+)
+
+file(COPY ${CMAKE_CURRENT_SOURCE_DIR}/conf DESTINATION ${CMAKE_CURRENT_BINARY_DIR}/)
+
+include(GoogleTest)
+gtest_discover_tests(gtest_utable)
+
+gtest_discover_tests(gtest_ipfix_exporter
+    TEST_LIST gtest_ipfix_exporter_tests
+)
+set_tests_properties(${gtest_ipfix_exporter_tests} PROPERTIES WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/)

deps/utable/test/conf/ipfix_ssl_test.json

@@ -0,0 +1,50 @@
+{
+    "flags_identify_info": [1,2,3],
+    "security_rule_list": ["aaa","bbb", "ccc"],
+    "tcp_rtt_ms": 61,
+    "ssl_ja3s_hash": "cd5a8d2e276eabf0839bf1a25acc479e",
+    "ssl_version": "v3",
+    "ssl_cn": "*.google.com",
+    "ssl_cert_issuer": "CN=GTS CA 1C3;O=Google Trust Services LLC;C=US;;;;",
+    "ssl_cert_subject": "CN=*.google.com;;;;;;",
+    "ssl_san": "*.google.com;*.appengine.google.com;*.bdn.dev;*.cloud.google.com;*.crowdsource.google.com;*.datacompute.google.com;*.google.ca;*.google.cl;*.google.co.in;*.google.co.jp;*.google.co.uk;*.google.com.ar;*.google.com.au;*.google.com.br;*.google.com.co;*.google.com.mx;*.google.com.tr;*.google.com.vn;*.google.de;*.google.es;*.google.fr;*.google.hu;*.google.it;*.google.nl;*.google.pl;*.google.pt;*.googleadapis.com;*.googleapis.cn;*.googlevideo.com;*.gstatic.cn;*.gstatic-cn.com;*.gstaticcnapps.cn;googlecnapps.cn;*.googlecnapps.cn;googleapps-cn.com;*.googleapps-cn.com;gkecnapps.cn;*.gkecnapps.cn;googledownloads.cn;*.googledownloads.cn;recaptcha.net.cn;*.recaptcha.net.cn;widevine.cn;*.widevine.cn;ampproject.org.cn;*.ampproject.org.cn;ampproject.net.cn;*.ampproject.net.cn;google-analytics-cn.com;*.google-analytics-cn.com;googleadservices-cn.com;*.googleadservices-cn.com;googlevads-cn.com;*.googlevads-cn.com;googleapis-cn.com;*.googleapis-cn.com;googleoptimize-cn.com;*.googleoptimize-cn.com;doubleclick-cn.net;*.doubleclick-cn.net;*.fls.doubleclick-cn.net;*.g.doubleclick-cn.net;doubleclick.cn;*.doubleclick.cn;*.fls.doubleclick.cn;*.g.doubleclick.cn;dartsearch-cn.net;*.dartsearch-cn.net;googletraveladservices-cn.com;*.googletraveladservices-cn.com;googletagservices-cn.com;*.googletagservices-cn.com;googletagmanager-cn.com;*.googletagmanager-cn.com;googlesyndication-cn.com;*.googlesyndication-cn.com;*.safeframe.googlesyndication-cn.com;app-measurement-cn.com;*.app-measurement-cn.com;gvt1-cn.com;*.gvt1-cn.com;gvt2-cn.com;*.gvt2-cn.com;2mdn-cn.net;*.2mdn-cn.net;googleflights-cn.net;*.googleflights-cn.net;admob-cn.com;*.admob-cn.com;*.gstatic.com;*.metric.gstatic.com;*.gvt1.com;*.gcpcdn.gvt1.com;*.gvt2.com;*.gcp.gvt2.com;*.url.google.com;*.youtube-nocookie.com;*.ytimg.com;android.com;*.android.com;*.flash.android.com;g.cn;*.g.cn;g.co;*.g.co;goo.gl;www.goo.gl;google-analytics.com;*.google-analytics.com;google.com;googlecommerce.com;*.googlecommerce.com;ggpht.cn",
+    "in_src_mac": "d4:c1:c8:8f:ac:f0",
+    "in_dest_mac": "d4:c1:c8:98:c7:60",
+    "tcp_client_isn": 2664633009,
+    "tcp_server_isn": 2183931785,
+    "address_type": 4,
+    "client_ip": "196.190.248.0",
+    "server_ip": "142.250.185.34",
+    "client_port": 16986,
+    "server_port": 443,
+    "in_link_id": 0,
+    "out_link_id": 0,
+    "start_timestamp_ms": 1702610615910,
+    "end_timestamp_ms": 1702610615996,
+    "duration_ms": 86,
+    "sent_pkts": 0,
+    "sent_bytes": 0,
+    "received_pkts": 10,
+    "received_bytes": 7246,
+    "tcp_c2s_ip_fragments": 0,
+    "tcp_s2c_ip_fragments": 0,
+    "tcp_c2s_rtx_pkts": 0,
+    "tcp_c2s_rtx_bytes": 0,
+    "tcp_s2c_rtx_pkts": 0,
+    "tcp_s2c_rtx_bytes": 0,
+    "tcp_c2s_o3_pkts": 0,
+    "tcp_s2c_o3_pkts": 0,
+    "tcp_c2s_lost_bytes": 0,
+    "tcp_s2c_lost_bytes": 0,
+    "flags": 0,
+    "decoded_as": "SSL",
+    "decoded_path": "ETHERNET.IPv4.UDP.VXLAN.ETHERNET.IPv4.TCP",
+    "t_vsys_id": 1,
+    "vsys_id": 1,
+    "session_id": 11849422307955,
+    "tcp_handshake_latency_ms": 37,
+    "server_os_desc": "Android",
+    "device_tag": "{\"tags\":[{\"tag\":\"device_id\",\"value\":\"device_1\"}]}",
+    "sled_ip": "127.0.0.1",
+    "dup_traffic_flag": 0
+}

deps/utable/test/unit_test_ipfix_exporter.cpp

@@ -0,0 +1,802 @@
+
+#include <stdio.h>
+#include <sys/time.h>
+#include <gtest/gtest.h>
+#include <netinet/in.h>
+#include "utable/utable.h"
+#include "cjson/cJSON.h"
+#include "uthash/utarray.h"
+
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+
+#define IPFIX_DEFUALT_VERSION 10
+#define GEEDGE_NETWORKS_PEN_NUMBER 54450
+#define IPFIX_DEFUALT_PEN_NUMBER GEEDGE_NETWORKS_PEN_NUMBER
+#define IPFIX_NONE_PEN_NUMBER -1
+#define IPFIX_TEMPLATE_SET_ID 2
+#define IPFIX_BUFF_MAX_SIZE 2048
+#define IPFIX_ELEMENT_MAX_LEN 32
+
+#ifndef MIN
+#define MIN(a, b) (((a) < (b)) ? (a) : (b))
+#endif
+
+enum ipfix_type
+{
+    IPFIX_UNSIGNED8 = 0,
+    IPFIX_UNSIGNED16,
+    IPFIX_UNSIGNED32,
+    IPFIX_UNSIGNED64,
+    IPFIX_VARIABLE_STRING,
+    IPFIX_UNKNOWN
+};
+
+struct ipfix_element
+{
+    enum ipfix_type type;
+    uint16_t element_id;
+    uint16_t value_length; // if type is variable_string, value_length is 0
+    int PEN_number;
+    char name[IPFIX_ELEMENT_MAX_LEN];
+};
+
+struct ipfix_template
+{
+    uint16_t template_id;
+    char *name;
+    UT_array *elements_array; // utarray for current template elements
+};
+
+struct ipfix_worker_context
+{
+    int source_id;
+    int sequence;
+    size_t template_blob_length;
+    char *template_blob;
+};
+
+struct ipfix_exporter_schema
+{
+    uint16_t n_worker;
+    uint16_t version;
+    int domain_id;
+    int PEN_number;
+    UT_array *templates_array;                         // utarray for templates
+    struct ipfix_worker_context *worker_context_array; // utarray for worker_context
+};
+
+struct ipfix_message_head
+{
+    uint16_t version;
+    uint16_t length;
+    int exporttime;
+    int ipfix_message_sequence;
+    int domain_id;
+};
+#define IPFIX_MESSAGE_HEAD_LEN sizeof(struct ipfix_message_head)
+
+#define THREAD_MAX 4
+#define TEMPLATE_MAX 13
+struct gtest_ipfix_template_info
+{
+    const char *template_name;
+    int template_id;
+    int n_elements;
+};
+
+struct gtest_ipfix_template_info gtest_ipfix_template_info[TEMPLATE_MAX] = {
+    {"BASE", 257, 87},
+    {"SSL", 258, 98},
+    {"HTTP", 259, 108},
+    {"MAIL", 260, 100},
+    {"DNS", 261, 106},
+    {"DTLS", 262, 97},
+    {"QUIC", 263, 90},
+    {"FTP", 264, 91},
+    {"SIP", 265, 102},
+    {"RTP", 266, 91},
+    {"SSH", 267, 98},
+    {"RDP", 268, 102},
+    {"Stratum", 269, 91}};
+
+#define IPFIX_ELEMENT_HEAD_LEN sizeof(uint16_t) + sizeof(uint16_t) + sizeof(uint32_t) // element_id + element_length + PEN_number
+const char *ipfix_schema_json_path = "./conf/ipfix_schema.json";
+int g_domain_id = 1;
+TEST(utable_ipfix_exporter_test, ipfix_schema_new_error_path)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new((const char *)"", g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema == NULL);
+}
+
+int test_template_blob_len_calculate(struct ipfix_exporter_schema *ipfix_schema)
+{
+    int template_blob_length = 0;
+    template_blob_length += IPFIX_MESSAGE_HEAD_LEN; // message head
+    template_blob_length += sizeof(uint16_t) + sizeof(uint16_t); // flow set id + flow set length
+    for (unsigned int i = 0; i < TEMPLATE_MAX; i++)
+    {
+        struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, i);
+        template_blob_length += sizeof(uint16_t) + sizeof(uint16_t) + utarray_len(template_item->elements_array) * (IPFIX_ELEMENT_HEAD_LEN); // sizeof(template_id) + sizeof(field_count) + n_elements * sizeof(element_id + element_length + PEN_number)
+    }
+
+    return template_blob_length;
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_schema_new_check_schema)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    EXPECT_EQ(ipfix_schema->domain_id, 1);
+    EXPECT_EQ(ipfix_schema->n_worker, THREAD_MAX);
+    EXPECT_EQ(utarray_len(ipfix_schema->templates_array), TEMPLATE_MAX);
+    EXPECT_EQ(ipfix_schema->PEN_number, GEEDGE_NETWORKS_PEN_NUMBER);
+    EXPECT_EQ(ipfix_schema->version, IPFIX_DEFUALT_VERSION);
+
+    for (unsigned int i = 0; i < TEMPLATE_MAX; i++)
+    {
+        struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, i);
+        ASSERT_TRUE(template_item != NULL);
+        EXPECT_EQ(template_item->template_id, gtest_ipfix_template_info[i].template_id);
+        EXPECT_EQ(gtest_ipfix_template_info[i].n_elements, utarray_len(template_item->elements_array));
+        EXPECT_STREQ(template_item->name, gtest_ipfix_template_info[i].template_name);
+    }
+
+    ASSERT_TRUE(ipfix_schema->worker_context_array != NULL);
+    for (int i = 0; i < THREAD_MAX; i++)
+    {
+        EXPECT_EQ(ipfix_schema->worker_context_array[i].source_id, (g_domain_id << 16) + i);
+        EXPECT_EQ(ipfix_schema->worker_context_array[i].sequence, 0);
+        ASSERT_TRUE(ipfix_schema->worker_context_array[i].template_blob != NULL);
+        EXPECT_EQ(ipfix_schema->worker_context_array[i].template_blob_length, test_template_blob_len_calculate(ipfix_schema));
+    }
+
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_template_flow_get0_all_elements_check_blob)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    for (int i = 0; i < THREAD_MAX; i++)
+    {
+        size_t blob_len = 0;
+        const char *blob = utable_ipfix_template_flow_get0(ipfix_schema, i, &blob_len);
+        ASSERT_TRUE(blob != NULL);
+        EXPECT_EQ(blob_len, test_template_blob_len_calculate(ipfix_schema));
+
+        // check header
+        struct ipfix_message_head *header = (struct ipfix_message_head *)blob;
+        struct ipfix_message_head header_value = {};
+        header_value.version = htons(header->version);
+        header_value.length = htons(header->length);
+        header_value.domain_id = htonl(header->domain_id);
+        header_value.ipfix_message_sequence = htonl(header->ipfix_message_sequence);
+        EXPECT_EQ(header_value.version, IPFIX_DEFUALT_VERSION);
+        EXPECT_EQ(header_value.length, test_template_blob_len_calculate(ipfix_schema));
+        EXPECT_EQ(header_value.domain_id, (g_domain_id << 16) + i);
+        EXPECT_EQ(header_value.ipfix_message_sequence, 0);
+
+        size_t offset = 0;
+        offset += IPFIX_MESSAGE_HEAD_LEN;
+
+        uint16_t flow_set_id = ntohs(*(uint16_t *)(blob + offset));
+        EXPECT_EQ(flow_set_id, IPFIX_TEMPLATE_SET_ID); // template set id
+        offset += 2;
+
+        uint16_t flow_set_length = ntohs(*(uint16_t *)(blob + offset));
+        EXPECT_EQ(flow_set_length, blob_len - IPFIX_MESSAGE_HEAD_LEN); // template set length
+        offset += 2;
+
+        for (unsigned int j = 0; j < TEMPLATE_MAX; j++)
+        {
+            uint16_t cur_template_blob_len = sizeof(uint16_t) + sizeof(uint16_t) + gtest_ipfix_template_info[j].n_elements * IPFIX_ELEMENT_HEAD_LEN;
+            ASSERT_LE(offset + cur_template_blob_len, blob_len);
+
+            uint16_t template_id = ntohs(*(uint16_t *)(blob + offset));
+            EXPECT_EQ(template_id, gtest_ipfix_template_info[j].template_id); // template id
+            offset += 2;
+
+            uint16_t field_count = ntohs(*(uint16_t *)(blob + offset));
+            EXPECT_EQ(field_count, gtest_ipfix_template_info[j].n_elements); // field count
+            offset += 2;
+
+            struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, j);
+            for (unsigned int p = 0; p < (unsigned int)gtest_ipfix_template_info[j].n_elements; p++)
+            {
+                struct ipfix_element *element = (struct ipfix_element *)utarray_eltptr(template_item->elements_array, p);
+                uint16_t PEN_number_flag = ntohs(*(uint16_t *)(blob + offset)) & 0x8000;
+                EXPECT_EQ(PEN_number_flag, 0x8000); // PEN number flag
+
+                uint16_t element_id = ntohs(*(uint16_t *)(blob + offset)) & 0x7fff;
+                EXPECT_EQ(element_id, element->element_id); // element id
+                offset += 2;
+
+                if (element->type == IPFIX_VARIABLE_STRING)
+                {
+                    uint16_t element_length = ntohs(*(uint16_t *)(blob + offset));
+                    EXPECT_EQ(element_length, 65535); // element length
+                }
+                else
+                {
+                    uint16_t element_length = ntohs(*(uint16_t *)(blob + offset));
+                    EXPECT_EQ(element_length, element->value_length); // element length
+                }
+                offset += 2;
+
+                uint32_t PEN_number = ntohl(*(uint32_t *)(blob + offset));
+                EXPECT_EQ(PEN_number, element->PEN_number); // PEN number
+                offset += 4;
+            }
+        }
+    }
+
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_template_flow_get0_all_elements_check_sequence)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    for (int i = 0; i < THREAD_MAX; i++)
+    {
+        size_t blob_len = 0;
+        const char *blob = utable_ipfix_template_flow_get0(ipfix_schema, i, &blob_len);
+        ASSERT_TRUE(blob != NULL);
+
+        // check header
+        struct ipfix_message_head *header = (struct ipfix_message_head *)blob;
+        struct ipfix_message_head header_value = {};
+        header_value.version = htons(header->version);
+        header_value.length = htons(header->length);
+        header_value.domain_id = htonl(header->domain_id);
+        header_value.ipfix_message_sequence = htonl(header->ipfix_message_sequence);
+        EXPECT_EQ(header_value.version, IPFIX_DEFUALT_VERSION);
+        EXPECT_EQ(header_value.length, test_template_blob_len_calculate(ipfix_schema));
+        EXPECT_EQ(header_value.domain_id, (g_domain_id << 16) + i);
+        EXPECT_EQ(header_value.ipfix_message_sequence, 0);
+        memset(&header_value, 0, sizeof(struct ipfix_message_head));
+        header = NULL;
+        blob = NULL;
+
+        // check sequence, utable_ipfix_template_flow_get0 will not increase sequence
+        blob = utable_ipfix_template_flow_get0(ipfix_schema, i, &blob_len);
+        ASSERT_TRUE(blob != NULL);
+
+        // check header
+        header = (struct ipfix_message_head *)blob;
+        header_value.ipfix_message_sequence = htonl(header->ipfix_message_sequence);
+        EXPECT_EQ(header_value.ipfix_message_sequence, 0);
+    }
+
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_template_get_not_found)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    int template_id = utable_ipfix_template_get(ipfix_schema, "test");
+    EXPECT_EQ(template_id, -1);
+
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_template_get_check_template_id)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    int template_id = 0;
+    for (unsigned int i = 0; i < TEMPLATE_MAX; i++)
+    {
+        template_id = utable_ipfix_template_get(ipfix_schema, gtest_ipfix_template_info[i].template_name);
+        EXPECT_EQ(template_id, i);
+
+        struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, i);
+        ASSERT_TRUE(template_item != NULL);
+    }
+
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+cJSON *test_ipifx_data_blob_to_cjson(struct ipfix_template *template_item, char *blob, size_t sz_blob)
+{
+    cJSON *root = cJSON_CreateObject();
+
+    size_t offset = IPFIX_MESSAGE_HEAD_LEN + sizeof(uint16_t) + sizeof(uint16_t); // skip message head and flow set id and flow set length
+    size_t n_elements = utarray_len(template_item->elements_array);
+    for (size_t i = 0; i < n_elements; i++)
+    {
+        struct ipfix_element *p_element = (struct ipfix_element *)utarray_eltptr(template_item->elements_array, i);
+        switch (p_element->type)
+        {
+            case IPFIX_UNSIGNED8:
+                {
+                    cJSON_AddNumberToObject(root, p_element->name, *(uint8_t *)(blob + offset));
+                    offset += sizeof(uint8_t);
+                    break;
+                }
+            case IPFIX_UNSIGNED16:
+                {
+                    cJSON_AddNumberToObject(root, p_element->name, ntohs(*(uint16_t *)(blob + offset)));
+                    offset += sizeof(uint16_t);
+                    break;
+                }
+            case IPFIX_UNSIGNED32:
+                {
+                    cJSON_AddNumberToObject(root, p_element->name, ntohl(*(uint32_t *)(blob + offset)));
+                    offset += sizeof(uint32_t);
+                    break;
+                }
+            case IPFIX_UNSIGNED64:
+                {
+                    cJSON_AddNumberToObject(root, p_element->name, be64toh(*(uint64_t *)(blob + offset)));
+                    offset += sizeof(uint64_t);
+                    break;
+                }
+            case IPFIX_VARIABLE_STRING:
+                {
+                    uint8_t sz_value = *(uint8_t *)(blob + offset);
+                    offset += sizeof(uint8_t);
+                    char *string_value = NULL;
+                    if (sz_value == 0)
+                    {
+                        break;
+                    }
+                    else if (sz_value < 255)
+                    {
+                        string_value = (char *)malloc(sizeof(char) * (sz_value + 1));
+                        memcpy(string_value, blob + offset, sz_value);
+                        string_value[sz_value] = '\0';
+                        offset += sz_value;
+                    }
+                    else // >= 255
+                    {
+                        uint16_t sz_long_string_value = ntohs(*(uint16_t *)(blob + offset));
+                        offset += sizeof(uint16_t);
+                        string_value = (char *)malloc(sizeof(char) * (sz_long_string_value + 1));
+                        memcpy(string_value, blob + offset, sz_long_string_value);
+                        string_value[sz_long_string_value] = '\0';
+                        offset += sz_long_string_value;
+                    }
+                    cJSON_AddStringToObject(root, p_element->name, string_value);
+                    free(string_value);
+                    break;
+                }
+            default:
+                break;
+        }
+    }
+
+    return root;
+}
+
+// after include before, and the extra part is cJSON_Number, but valueint = 0;
+void test_ipfix_compare_cjson(cJSON *before, cJSON *after)
+{
+    // find before's element in after
+    cJSON *item = before->child;
+    while (item != NULL)
+    {
+        if (item->type == cJSON_Number)
+        {
+            cJSON *item_after = cJSON_GetObjectItem(after, item->string);
+            ASSERT_TRUE(item_after != NULL);
+            EXPECT_EQ(item->valueint, item_after->valueint);
+        }
+        if (item->type == cJSON_String)
+        {
+            cJSON *item_after = cJSON_GetObjectItem(after, item->string);
+            ASSERT_TRUE(item_after != NULL);
+            EXPECT_STREQ(item->valuestring, item_after->valuestring);
+        }
+        item = item->next;
+    }
+
+    // find after's element in before
+    item = after->child;
+    while (item != NULL)
+    {
+        if (item->type == cJSON_Number) // if after's element is cJSON_Number, before's element may be empty or cJSON_Number
+        {
+            cJSON *item_before = cJSON_GetObjectItem(before, item->string);
+            if (item_before == NULL) // if before's element is empty, after's element valueint must be 0
+            {
+                EXPECT_EQ(item->valueint, 0);
+            }
+            else
+            {
+                EXPECT_EQ(item->valueint, item_before->valueint); // if before's element is cJSON_Number, after's element valueint must be equal to before's element valueint
+            }
+        }
+        if (item->type == cJSON_String)
+        {
+            cJSON *item_before = cJSON_GetObjectItem(before, item->string);
+            ASSERT_TRUE(item_before != NULL);
+            EXPECT_STREQ(item->valuestring, item_before->valuestring);
+        }
+        item = item->next;
+    }
+}
+
+int test_empty_data_blob_len_calculate(struct ipfix_exporter_schema *ipfix_schema, int template_id)
+{
+    struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, (unsigned int)template_id);
+    int data_blob_length = 0;
+    data_blob_length += IPFIX_MESSAGE_HEAD_LEN;                                   // message head
+    data_blob_length += sizeof(uint16_t) + sizeof(uint16_t);                      // flow set id + flow set length
+    for (unsigned int i = 0; i < utarray_len(template_item->elements_array); i++)
+    {
+        struct ipfix_element *element = (struct ipfix_element *)utarray_eltptr(template_item->elements_array, i);
+        if (element->type == IPFIX_VARIABLE_STRING)
+        {
+            data_blob_length += 1;
+        }
+        else
+        {
+            data_blob_length += element->value_length; // element length
+        }
+    }
+
+    return data_blob_length;
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_data_flow_empty_utable_check_blob)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    for (unsigned int i = 0; i < TEMPLATE_MAX; i++)
+    {
+        cJSON *before = cJSON_CreateObject();
+        struct utable *table = utable_new();
+        int template_id = utable_ipfix_template_get(ipfix_schema, gtest_ipfix_template_info[i].template_name);
+        ASSERT_EQ(template_id, i);
+
+        for (unsigned int j = 0; j < THREAD_MAX; j++)
+        {
+            size_t blob_len = 0;
+            char *blob = NULL;
+            utable_ipfix_data_flow_exporter(table, ipfix_schema, template_id, j, &blob, &blob_len);
+            ASSERT_TRUE(blob != NULL);
+
+            struct ipfix_message_head *header = (struct ipfix_message_head *)blob;
+            struct ipfix_message_head header_value = {};
+            header_value.version = htons(header->version);
+            header_value.length = htons(header->length);
+            header_value.domain_id = htonl(header->domain_id);
+            header_value.ipfix_message_sequence = htonl(header->ipfix_message_sequence);
+
+            ASSERT_EQ(header_value.version, IPFIX_DEFUALT_VERSION);
+            ASSERT_EQ(header_value.length, blob_len);
+            ASSERT_EQ(header_value.length, test_empty_data_blob_len_calculate(ipfix_schema, i));
+            ASSERT_EQ(header_value.domain_id, (g_domain_id << 16) + j);
+            ASSERT_EQ(header_value.ipfix_message_sequence, i);
+
+            cJSON *after = test_ipifx_data_blob_to_cjson((struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, i), blob, blob_len);
+            test_ipfix_compare_cjson(before, after);
+            cJSON_Delete(after);
+            free(blob);
+            blob = NULL;
+        }
+
+        cJSON_Delete(before);
+        utable_free(table);
+    }
+
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_data_flow_utable_value_type_integer)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    cJSON *before = cJSON_CreateObject();
+    struct utable *table = utable_new();
+    utable_add_integer(table, "session_id", 123456789);
+    cJSON_AddNumberToObject(before, "session_id", 123456789);
+    
+    unsigned int template_id = utable_ipfix_template_get(ipfix_schema, "HTTP");
+    char *blob = NULL;
+    size_t sz_blob = 0;
+    utable_ipfix_data_flow_exporter(table, ipfix_schema, template_id, 0, &blob, &sz_blob);
+    ASSERT_EQ(sz_blob, test_empty_data_blob_len_calculate(ipfix_schema, template_id)); // integer value length has been calculated in test_empty_data_blob_len_calculate
+
+    cJSON *after = test_ipifx_data_blob_to_cjson((struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, template_id), blob, sz_blob);
+    test_ipfix_compare_cjson(before, after);
+
+    cJSON_Delete(before);
+    cJSON_Delete(after);
+    utable_free(table);
+    free(blob);
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_data_flow_utable_value_type_cstring)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    cJSON *before = cJSON_CreateObject();
+    struct utable *table = utable_new();
+    utable_add_integer(table, "session_id", 123456789);
+    cJSON_AddNumberToObject(before, "session_id", 123456789);
+
+    utable_add_cstring(table, "http_url", "http://www.baidu.com", strlen("http://www.baidu.com"));
+    cJSON_AddStringToObject(before, "http_url", "http://www.baidu.com");
+    
+    unsigned int template_id = utable_ipfix_template_get(ipfix_schema, "HTTP");
+    char *blob = NULL;
+    size_t sz_blob = 0;
+    utable_ipfix_data_flow_exporter(table, ipfix_schema, template_id, 0, &blob, &sz_blob);
+    ASSERT_EQ(sz_blob, test_empty_data_blob_len_calculate(ipfix_schema, template_id) + strlen("http://www.baidu.com")); // variable_string length need to be calculated
+
+    cJSON *after = test_ipifx_data_blob_to_cjson((struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, template_id), blob, sz_blob);
+    test_ipfix_compare_cjson(before, after);
+
+    cJSON_Delete(before);
+    cJSON_Delete(after);
+    utable_free(table);
+    free(blob);
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+char http_url_test[] = "storeapi.app-vtion.com/storeApi/systemApp/checkAppStatus.json?store=com.tcl.appmarket2&mac=3C:59:1E:DC:F8:F4&accessKeyId=1472540256617";
+TEST(utable_ipfix_exporter_test, ipfix_data_flow_variable_string_6000) // 6000 bytes > 2 * IPFIX_BUFF_MAX_SIZE, test variable_string length > 255 and realloc
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    char *http_url = NULL;
+    struct utable *table = utable_new();
+    cJSON *before = cJSON_CreateObject();
+    utable_add_cstring(table, "decoded_as", "HTTP", strlen("HTTP"));
+    cJSON_AddStringToObject(before, "decoded_as", "HTTP");
+
+    http_url = (char *)malloc(sizeof(char) * 7000);
+
+    int offset = 0;
+    while (offset < 6000)
+    {
+        memcpy(http_url + offset, http_url_test, sizeof(http_url_test));
+        offset += sizeof(http_url_test);
+    }
+    utable_add_cstring(table, "http_url", http_url, offset);
+    cJSON_AddStringToObject(before, "http_url", http_url);
+
+    unsigned int template_id = utable_ipfix_template_get(ipfix_schema, "HTTP");
+    char *blob = NULL;
+    size_t sz_blob = 0;
+    utable_ipfix_data_flow_exporter(table, ipfix_schema, template_id, 0, &blob, &sz_blob);
+    ASSERT_EQ(sz_blob, test_empty_data_blob_len_calculate(ipfix_schema, template_id) + offset + strlen("HTTP") + 2); // variable_string length need to be calculated, if length > 255, length + 2
+
+    cJSON *after = test_ipifx_data_blob_to_cjson((struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, template_id), blob, sz_blob);
+    test_ipfix_compare_cjson(before, after);
+
+    cJSON_Delete(before);
+    cJSON_Delete(after);
+    utable_free(table);
+    free(blob);
+    free(http_url);
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+TEST(utable_ipfix_exporter_test, ipfix_data_flow_utable_value_type_integer_array)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    cJSON *before = cJSON_CreateObject();
+    struct utable *table = utable_new();
+    utable_add_integer(table, "session_id", 123456789);
+    cJSON_AddNumberToObject(before, "session_id", 123456789);
+
+    int64_t monitor_rule_list[8] = {1, 2, 3, 4, 5, 6, 7, 8};
+    utable_add_integer_array(table, "monitor_rule_list", monitor_rule_list, 8);
+    cJSON_AddStringToObject(before, "monitor_rule_list", "[1,2,3,4,5,6,7,8]");
+    
+    unsigned int template_id = utable_ipfix_template_get(ipfix_schema, "HTTP");
+    char *blob = NULL;
+    size_t sz_blob = 0;
+    utable_ipfix_data_flow_exporter(table, ipfix_schema, template_id, 0, &blob, &sz_blob);
+    ASSERT_EQ(sz_blob, test_empty_data_blob_len_calculate(ipfix_schema, template_id) + strlen("[1,2,3,4,5,6,7,8]")); // variable_string length need to be calculated
+
+    cJSON *after = test_ipifx_data_blob_to_cjson((struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, template_id), blob, sz_blob);
+    test_ipfix_compare_cjson(before, after);
+
+    cJSON_Delete(before);
+    cJSON_Delete(after);
+    utable_free(table);
+    free(blob);
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+extern "C" int load_file_to_memory(const char *file_name, unsigned char **pp_out, size_t *out_sz);
+cJSON *ipfix_init_utable_from_log_json(struct utable *table, const char *test_json_path, size_t *sz_blob)
+{
+    size_t json_size = 0;
+    unsigned char *json_str = NULL;
+    load_file_to_memory(test_json_path, &json_str, &json_size);
+    if (json_str == NULL || json_size == 0)
+    {
+        return NULL;
+    }
+
+    cJSON *root = NULL;
+    root = cJSON_Parse((const char *)json_str);
+    cJSON *item = root->child;
+    while (item != NULL)
+    {
+        switch (item->type)
+        {
+            case cJSON_Number:
+            {
+                utable_add_integer(table, item->string, item->valueint);
+                break;
+            }
+            case cJSON_String:
+            {
+                int sz_value = strlen(item->valuestring);
+                utable_add_cstring(table, item->string, item->valuestring, sz_value);
+                if (sz_value < 255)
+                {
+                    *sz_blob += strlen(item->valuestring);
+                }
+                else
+                {
+                    *sz_blob += (strlen(item->valuestring) + 2);
+                }
+                break;
+            }
+            case cJSON_Array: 
+            {
+                cJSON *array_item = cJSON_GetArrayItem(item, 0);
+                size_t array_sz = cJSON_GetArraySize(item);
+                if (array_sz > 0)
+                {
+                    if (array_item->type == cJSON_Number)
+                    {
+                        int *array = (int *)malloc(sizeof(int) * array_sz);
+                        int64_t *long_array = (int64_t *)malloc(sizeof(int64_t) * array_sz);
+                        for (size_t i = 0; i < array_sz; i++)
+                        {
+                            array_item = cJSON_GetArrayItem(item, i);
+                            if (array_item->type == cJSON_Number)
+                            {
+                                array[i] = array_item->valueint;
+                                long_array[i] = array_item->valueint;
+                            }
+                        }
+                        utable_add_integer_array(table, item->string, long_array, array_sz);
+
+                        cJSON *json_array = cJSON_CreateIntArray((const int *)array, array_sz);
+                        cJSON_Delete(item->child);
+                        item->type = cJSON_String;
+                        item->child = NULL;
+                        item->valuestring = cJSON_PrintUnformatted(json_array);
+                        cJSON_Delete(json_array);
+
+                        free(array);
+                        free(long_array);
+                    }
+                    if (array_item->type == cJSON_String)
+                    {
+                        char **array = (char **)malloc(sizeof(char *) * array_sz);
+                        size_t *sz_value = (size_t *)malloc(sizeof(size_t) * array_sz);
+                        for (size_t i = 0; i < array_sz; i++)
+                        {
+                            array_item = cJSON_GetArrayItem(item, i);
+                            if (array_item->type == cJSON_String)
+                            {
+                                sz_value[i] = strlen(array_item->valuestring);
+                                array[i] = strdup(array_item->valuestring);
+                            }
+                        }
+                        utable_add_cstring_array(table, item->string, (const char **)array, sz_value, array_sz);
+
+                        cJSON *json_array = cJSON_CreateStringArray((const char **)array, array_sz);
+                        item->valuestring = cJSON_PrintUnformatted(json_array);
+                        item->type = cJSON_String;
+                        cJSON_Delete(item->child);
+                        item->child = NULL;
+                        cJSON_Delete(json_array);
+
+                        for (size_t i = 0; i < array_sz; i++)
+                        {
+                            free(array[i]);
+                        }
+                        free(array);
+                    }
+                    int sz_value = strlen(item->valuestring);
+                    if (sz_value < 255)
+                    {
+                        *sz_blob += sz_value;
+                    }
+                    else
+                    {
+                        *sz_blob += (sz_value + 2);
+                    }
+                }
+                break;
+            }
+            default:
+                break;
+        }
+        item = item->next;
+    }
+    free(json_str);
+    return root;
+}
+
+const char *ssl_test_json_path = "./conf/ipfix_ssl_test.json";
+TEST(utable_ipfix_exporter_test, ipfix_data_flow_ssl_log_test)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    struct utable *table = utable_new();
+    unsigned int template_id = utable_ipfix_template_get(ipfix_schema, "SSL");
+    size_t predict_sz_blob = test_empty_data_blob_len_calculate(ipfix_schema, template_id);
+
+    cJSON *before_json_root = ipfix_init_utable_from_log_json(table, ssl_test_json_path, &predict_sz_blob);
+
+    char *blob = NULL;
+    size_t sz_blob = 0;
+    utable_ipfix_data_flow_exporter(table, ipfix_schema, template_id, 0, &blob, &sz_blob);
+    ASSERT_EQ(sz_blob, predict_sz_blob);
+
+    cJSON *after_json_root = test_ipifx_data_blob_to_cjson((struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, template_id), blob, sz_blob);
+
+    char *before_json_str = cJSON_Print(before_json_root);
+    char *after_json_str = cJSON_Print(after_json_root);
+    printf("before_json: %s\n", before_json_str);
+    printf("after_json: %s\n", after_json_str);
+    free(before_json_str);
+    free(after_json_str);
+
+    test_ipfix_compare_cjson(before_json_root, after_json_root);
+
+    cJSON_Delete(before_json_root);
+    cJSON_Delete(after_json_root);
+    free(blob);
+    utable_free(table);
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+const char *http_test_json_path = "./conf/ipfix_http_test.json";
+TEST(utable_ipfix_exporter_test, ipfix_data_flow_http_log_test)
+{
+    struct ipfix_exporter_schema *ipfix_schema = utable_ipfix_exporter_schema_new(ipfix_schema_json_path, g_domain_id, THREAD_MAX);
+    ASSERT_TRUE(ipfix_schema != NULL);
+
+    struct utable *table = utable_new();
+    unsigned int template_id = utable_ipfix_template_get(ipfix_schema, "HTTP");
+    size_t predict_sz_blob = test_empty_data_blob_len_calculate(ipfix_schema, template_id);
+
+    cJSON *before_json_root = ipfix_init_utable_from_log_json(table, http_test_json_path, &predict_sz_blob);
+
+    char *blob = NULL;
+    size_t sz_blob = 0;
+    utable_ipfix_data_flow_exporter(table, ipfix_schema, template_id, 0, &blob, &sz_blob);
+    ASSERT_EQ(sz_blob, predict_sz_blob);
+
+    cJSON *after_json_root = test_ipifx_data_blob_to_cjson((struct ipfix_template *)utarray_eltptr(ipfix_schema->templates_array, template_id), blob, sz_blob);
+    test_ipfix_compare_cjson(before_json_root, after_json_root);
+
+    cJSON_Delete(before_json_root);
+    cJSON_Delete(after_json_root);
+    free(blob);
+    utable_free(table);
+    utable_ipfix_exporter_schema_free(ipfix_schema);
+}
+
+int main(int argc, char *argv[])
+{
+    testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

deps/utable/test/unit_test_utable.cpp

@@ -0,0 +1,773 @@
+
+#include <gtest/gtest.h>
+
+#include "base64/b64.h"
+#include "utable/utable.h"
+#include "cjson/cJSON.h"
+#include "mpack/mpack.h"
+
+
+
+void test_utable_assert_arr(const struct utable *table, const char *key, int64_t expected_arr[], size_t n_expected_arr)
+{
+    int64_t *arr = NULL;
+    size_t n_arr = 0;
+    EXPECT_EQ(utable_get_value_type(table, key), utable_value_type_integer_array);
+    utable_get0_integer_value_array(table, key, &arr, &n_arr);
+    EXPECT_EQ(n_arr, n_expected_arr);
+    for (size_t i = 0; i < n_arr; i++) {
+        EXPECT_EQ(expected_arr[i], arr[i]);
+    }
+}
+
+void test_utable_assert_str(const struct utable *table, const char *key, const char *expected_str)
+{
+    char *str = NULL;
+    size_t str_len = 0;
+    EXPECT_EQ(utable_get_value_type(table, key), utable_value_type_cstring);
+    utable_get0_cstring_value(table, key, &str, &str_len);
+    EXPECT_EQ(str_len, strlen(expected_str));
+    EXPECT_EQ(strncmp(str, expected_str, str_len), 0);
+}
+
+void test_utable_assert_str_array(const struct utable *table, const char *key, const char **expected_str_array, size_t n_expected_str)
+{
+    char **str_array = NULL;
+    size_t *str_len = NULL;
+    size_t n_str = 0;
+    EXPECT_EQ(utable_get_value_type(table, key), utable_value_type_cstring_array);
+    utable_get0_cstring_value_array(table, key, &str_array, &str_len, &n_str);
+    EXPECT_EQ(n_str, n_expected_str);
+    for (size_t i = 0; i < n_str; i++) {
+        EXPECT_EQ(str_len[i], strlen(expected_str_array[i]));
+        EXPECT_EQ(strncmp(str_array[i], expected_str_array[i], str_len[i]), 0);
+    }
+}
+
+void test_utable_assert_blob(const struct utable *table, const char *key, const char *expected_blob, size_t expected_blob_len)
+{
+    char *blob = NULL;
+    size_t blob_len = 0;
+    EXPECT_EQ(utable_get_value_type(table, key), utable_value_type_blob);
+    utable_get0_blob_value(table, key, &blob, &blob_len);
+    EXPECT_EQ(blob_len, expected_blob_len);
+    EXPECT_EQ(memcmp(blob, expected_blob, blob_len), 0);
+}
+
+TEST(utable_test, new_del_when_empty)
+{
+    struct utable *table = utable_new();
+    ASSERT_TRUE(table != NULL);
+    utable_free(table);
+}
+
+TEST(utable_test, add_1_int_and_query)
+{
+    struct utable *table = utable_new();
+    int64_t int_array[] = { 1 };
+    utable_add_integer_array(table, "key", int_array, 1);
+
+    test_utable_assert_arr(table, "key", int_array, 1);
+    
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+sizeof(int64_t);
+    B.n_integer_array=1;
+    B.n_integer_array_size=sizeof(int64_t);
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+    
+    utable_free(table);
+}
+
+TEST(utable_test, add_2_int_and_query)
+{
+    struct utable *table = utable_new();
+    int64_t int_array[] = { 1, 2 };
+    utable_add_integer_array(table, "key", int_array, 2);
+    test_utable_assert_arr(table, "key", int_array, 2);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+sizeof(int64_t)*2;
+    B.n_integer_array=1;
+    B.n_integer_array_size=sizeof(int64_t)*2;
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+}
+
+TEST(utable_test, add_string_and_query)
+{
+    struct utable *table = utable_new();
+    const char *str = "hello world";
+    utable_add_cstring(table, "key", str, strlen(str));
+    test_utable_assert_str(table, "key", str);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+strlen(str);
+    B.n_cstring=1;
+    B.n_cstring_size=strlen(str);
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+}
+
+TEST(utable_test, add_string_array_and_query)
+{
+    struct utable *table = utable_new();
+    const char *str[] = {"hello world", "foo bar"};
+    size_t n_str=sizeof(str)/sizeof(str[0]);
+    size_t str_sz[] = {strlen(str[0]), strlen(str[1])};
+    utable_add_cstring_array(table, "key", str, str_sz, n_str);
+    test_utable_assert_str_array(table, "key", str, n_str);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+str_sz[0]+str_sz[1];
+    B.n_cstring_array=1;
+    B.n_cstring_array_size=str_sz[0]+str_sz[1];
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+}
+
+TEST(utable_test, add_blob_and_query)
+{
+    struct utable *table = utable_new();
+    const char *blob = "hello world";
+    size_t blob_len = strlen(blob);
+    utable_add_blob(table, "key", blob, blob_len);
+    test_utable_assert_blob(table, "key", blob, blob_len);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+blob_len;
+    B.n_blob=1;
+    B.n_blob_size=blob_len;
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+}
+
+void test_utable_check_if_value_in_json(cJSON *tag_obj, const char *key, const std::string &value)
+{
+    cJSON *tag_val = cJSON_GetObjectItem(tag_obj, key);
+    EXPECT_NE(tag_val, nullptr);
+    EXPECT_STREQ(tag_val->valuestring, value.c_str());
+}
+
+void test_utable_check_if_value_in_json(cJSON *tag_obj, const char *key, const char *value[], size_t n_value)
+{
+    cJSON *tag_val = cJSON_GetObjectItem(tag_obj, key);
+    EXPECT_NE(tag_val, nullptr);
+    EXPECT_EQ(tag_val->type, cJSON_Array);
+    int arr_size = cJSON_GetArraySize(tag_val);
+    EXPECT_EQ(arr_size, n_value);
+    for (int i = 0; i < arr_size; i++) {
+        cJSON *arr_item = cJSON_GetArrayItem(tag_val, i);
+        EXPECT_STREQ(arr_item->valuestring, value[i]);
+    }
+}
+
+void test_utable_check_if_value_in_json(cJSON *tag_obj, const char *key, int64_t value[], size_t n_value)
+{
+    cJSON *tag_val = cJSON_GetObjectItem(tag_obj, key);
+    EXPECT_NE(tag_val, nullptr);
+    EXPECT_EQ(tag_val->type, cJSON_Array);
+    int arr_size = cJSON_GetArraySize(tag_val);
+    EXPECT_EQ(arr_size, n_value);
+    for (int i = 0; i < arr_size; i++) {
+        cJSON *arr_item = cJSON_GetArrayItem(tag_val, i);
+        EXPECT_EQ(arr_item->valueint, value[i]);
+    }
+}
+
+void test_utable_check_if_value_in_json(cJSON *tag_obj, const char *key, const char *value, size_t blob_len)
+{
+    cJSON *tag_val = cJSON_GetObjectItem(tag_obj, key);
+    EXPECT_NE(tag_val, nullptr);
+    // the tag_val is encoded in base64
+    size_t dec_size = 0;
+    unsigned char *dec = b64_decode_ex(tag_val->valuestring, strlen(tag_val->valuestring), &dec_size);
+    EXPECT_EQ(dec_size, blob_len);
+    EXPECT_EQ(memcmp(dec, value, blob_len), 0);
+
+    free(dec);
+}
+
+TEST(utable_test, export_json)
+{
+    struct utable *table = utable_new();
+    int64_t int_array[] = { 1, 2 };
+    const char *str = "hello world";
+    const char *blob = "bin hello world";
+
+    const char *str_array[] = {"hello world", "foo bar"};
+    size_t n_str=sizeof(str_array)/sizeof(str_array[0]);
+    size_t str_sz[] = {strlen(str_array[0]), strlen(str_array[1])};
+
+    utable_add_integer_array(table, "key1", int_array, 2);
+    utable_add_cstring(table, "key2", str, strlen(str));
+    utable_add_blob(table, "key3", blob, strlen(blob));
+    utable_add_cstring_array(table, "key4", str_array, str_sz, n_str);
+    
+    char *json = NULL;
+    size_t json_len = 0;
+    EXPECT_EQ(utable_json_export(table, &json, &json_len), 0);
+    EXPECT_NE(nullptr, json); 
+    cJSON *root = cJSON_Parse(json);
+    test_utable_check_if_value_in_json(root, "key1", int_array, 2);
+    std::string expected_str(str);
+    test_utable_check_if_value_in_json(root, "key2", expected_str);
+    test_utable_check_if_value_in_json(root, "key3", blob, strlen(blob));
+
+    test_utable_check_if_value_in_json(root, "key4", str_array, n_str);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=4;
+    B.n_item_size=strlen("key1")+sizeof(int64_t)*2+strlen("key2")+strlen(str)+strlen("key3")+strlen(blob)+strlen("key4")+str_sz[0]+str_sz[1];
+    B.n_blob=1;
+    B.n_blob_size=strlen(blob);
+    B.n_cstring=1;
+    B.n_cstring_size=strlen(str);
+    B.n_integer_array=1;
+    B.n_integer_array_size=sizeof(int64_t)*2;
+    B.n_integer=0;
+    B.n_integer_size=0;
+    B.n_cstring_array=1;
+    B.n_cstring_array_size=str_sz[0]+str_sz[1];
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+    free(json);
+    cJSON_Delete(root);
+}
+
+
+TEST(utable_test, export_invalid_unicode_string_to_json)
+{
+    struct utable *table = utable_new();
+    const char *str = "googleads.g.doublec\x86rck.net";
+    utable_add_cstring(table, "key2", str, strlen(str));
+    
+    char *json = NULL;
+    size_t json_len = 0;
+    EXPECT_EQ(utable_json_export(table, &json, &json_len), 0);
+    EXPECT_NE(nullptr, json); 
+    cJSON *root = cJSON_Parse(json);
+    std::string expected_str(str);
+    test_utable_check_if_value_in_json(root, "key2", expected_str);
+    
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key2")+strlen(str);
+    B.n_cstring=1;
+    B.n_cstring_size=strlen(str);
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+    free(json);
+    cJSON_Delete(root);
+}
+
+char *test_get_cstring_from_mpack_node(mpack_node_t node)
+{
+    size_t len = mpack_node_strlen(node);
+    char *str = (char *)malloc(len + 1);
+    memcpy(str, mpack_node_str(node), len);
+    str[len] = '\0';
+    return str;
+}
+
+void test_utable_check_if_value_in_mpack(mpack_node_t item, const char *key, const std::string &value)
+{
+    mpack_node_t key_node = mpack_node_map_cstr(item, "key");
+    mpack_node_t type_node = mpack_node_map_cstr(item, "type");
+    mpack_node_t value_node = mpack_node_map_cstr(item, "value");
+
+    char *key_tmp = test_get_cstring_from_mpack_node(key_node);
+    char *type_tmp = test_get_cstring_from_mpack_node(type_node);
+    char *value_tmp = test_get_cstring_from_mpack_node(value_node);
+    EXPECT_STREQ(key_tmp, key);
+    EXPECT_STREQ(type_tmp, "c_str");
+    EXPECT_STREQ(value_tmp, value.c_str());
+
+    free(key_tmp);
+    free(type_tmp);
+    free(value_tmp);
+}
+
+void test_utable_check_if_value_in_mpack(mpack_node_t item, const char *key, int64_t value[], size_t n_value)
+{
+    mpack_node_t key_node = mpack_node_map_cstr(item, "key");
+    mpack_node_t type_node = mpack_node_map_cstr(item, "type");
+    mpack_node_t value_node = mpack_node_map_cstr(item, "value");
+    char *key_tmp = test_get_cstring_from_mpack_node(key_node);
+    char *type_tmp = test_get_cstring_from_mpack_node(type_node);
+
+    EXPECT_STREQ(key_tmp, key);
+    EXPECT_STREQ(type_tmp, "i_arr");
+    EXPECT_EQ(mpack_node_array_length(value_node), n_value);
+    for (size_t i = 0; i < n_value; i++) {
+        mpack_node_t tmp_v_node = mpack_node_array_at(value_node, i);
+        EXPECT_EQ(mpack_node_i64(tmp_v_node), value[i]);
+    }
+    free(key_tmp);
+    free(type_tmp);
+}
+
+void test_utable_check_if_value_in_mpack(mpack_node_t item, const char *key, const char *value, size_t blob_len)
+{
+    mpack_node_t key_node = mpack_node_map_cstr(item, "key");
+    mpack_node_t type_node = mpack_node_map_cstr(item, "type");
+    mpack_node_t value_node = mpack_node_map_cstr(item, "value");
+    char *key_tmp = test_get_cstring_from_mpack_node(key_node);
+    char *type_tmp = test_get_cstring_from_mpack_node(type_node);
+    EXPECT_STREQ(key_tmp, key);
+    EXPECT_STREQ(type_tmp, "blob");
+    EXPECT_EQ(mpack_node_bin_size(value_node), blob_len);
+    EXPECT_EQ(memcmp(mpack_node_bin_data(value_node), value, blob_len), 0);
+
+    free(key_tmp);
+    free(type_tmp);
+}
+
+TEST(utable_test, DISABLED_export_mpack)
+{
+    struct utable *table = utable_new();
+    int64_t int_array[] = { 1, 2 };
+    const char *str = "hello world";
+    const char *blob = "bin hello world";
+    utable_add_integer_array(table, "key1", int_array, 2);
+    utable_add_cstring(table, "key2", str, strlen(str));
+    utable_add_blob(table, "key3", blob, strlen(blob));
+
+    char *mpack = NULL;
+    size_t mpack_len = 0;
+    EXPECT_EQ(utable_msgpack_export(table, &mpack, &mpack_len), 0);
+
+
+    mpack_tree_t tree;
+    mpack_tree_init_data(&tree, mpack, mpack_len);
+    mpack_tree_parse(&tree);
+    mpack_node_t root = mpack_tree_root(&tree);
+    size_t n_item = mpack_node_array_length(root);
+    EXPECT_EQ(n_item, 3);
+
+    test_utable_check_if_value_in_mpack(mpack_node_array_at(root, 0), "key1", int_array, 2);
+    std::string expected_str(str);
+    test_utable_check_if_value_in_mpack(mpack_node_array_at(root, 1), "key2", expected_str);
+    test_utable_check_if_value_in_mpack(mpack_node_array_at(root, 2), "key3", blob, strlen(blob));
+
+    utable_free(table);
+    free(mpack);
+    mpack_tree_destroy(&tree);
+}
+
+
+TEST(utable_test, add_blob_then_del)
+{
+    struct utable *table = utable_new();
+    const char *blob = "hello world";
+    size_t blob_len = strlen(blob);
+    utable_add_blob(table, "key", blob, blob_len);
+    test_utable_assert_blob(table, "key", blob, blob_len);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+strlen(blob);
+    B.n_blob=1;
+    B.n_blob_size=strlen(blob);
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    char *value;
+    size_t value_len;
+    utable_delete_item(table, "key");
+    EXPECT_EQ(utable_get0_blob_value(table, "key", &value, &value_len), -1);
+    EXPECT_EQ(utable_get0_cstring_value(table, "key", &value, &value_len), -1);
+    int64_t *value_array;
+    size_t n_value;
+    EXPECT_EQ(utable_get0_integer_value_array(table, "key", &value_array, &n_value), -1);    
+    EXPECT_EQ(utable_get_value_type(table, "key"), utable_value_type_undefined);
+
+    struct utable_stat C={};
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &C, sizeof(struct utable_stat)), 0);
+    
+    utable_free(table);
+}
+
+
+TEST(utable_test, add_string_array_then_del)
+{
+    struct utable *table = utable_new();
+    const char *str[] = {"hello world", "foo bar"};
+    size_t n_str=sizeof(str)/sizeof(str[0]);
+    size_t str_sz[] = {strlen(str[0]), strlen(str[1])};
+    utable_add_cstring_array(table, "key", str, str_sz, n_str);
+    test_utable_assert_str_array(table, "key", str, n_str);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+str_sz[0]+str_sz[1];
+    B.n_cstring_array=1;
+    B.n_cstring_array_size=str_sz[0]+str_sz[1];
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    char *value;
+    size_t value_len;
+    utable_delete_item(table, "key");
+    EXPECT_EQ(utable_get0_blob_value(table, "key", &value, &value_len), -1);
+    EXPECT_EQ(utable_get0_cstring_value(table, "key", &value, &value_len), -1);
+    int64_t *value_array;
+    size_t n_value;
+    EXPECT_EQ(utable_get0_integer_value_array(table, "key", &value_array, &n_value), -1);    
+
+    char **str_array = NULL;
+    size_t *str_len = NULL;
+    n_str = 0;
+    EXPECT_EQ(utable_get0_cstring_value_array(table, "key", &str_array, &str_len, &n_str), -1); 
+
+    EXPECT_EQ(utable_get_value_type(table, "key"), utable_value_type_undefined);
+
+    struct utable_stat C={};
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &C, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+}
+
+TEST(utable_test, query_on_wrong_key)
+{
+    struct utable *table = utable_new();
+    
+    char *value;
+    size_t value_len;
+    EXPECT_EQ(utable_get0_blob_value(table, "key", &value, &value_len), -1);
+    EXPECT_EQ(utable_get0_cstring_value(table, "key", &value, &value_len), -1);
+    int64_t *value_array;
+    size_t n_value;
+    EXPECT_EQ(utable_get0_integer_value_array(table, "key", &value_array, &n_value), -1);
+    EXPECT_EQ(utable_get_value_type(table, "key"), utable_value_type_undefined);
+
+    utable_free(table);
+}
+
+TEST(utable_test, add_on_dup_key)
+{
+    struct utable *table = utable_new();
+    int64_t int_array[] = { 1, 2 };
+    const char *str = "hello world";
+    utable_add_integer_array(table, "key", int_array, 2);
+    utable_add_cstring(table, "key", str, strlen(str));
+    
+    test_utable_assert_arr(table, "key", int_array, 2);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key")+sizeof(int64_t)*2;
+    B.n_integer_array=1;
+    B.n_integer_array_size=sizeof(int64_t)*2;
+
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+}
+
+TEST(utable_test, iter_and_query)
+{
+    struct utable *table = utable_new();
+    
+    int64_t int_array[] = { 1, 2 };
+    const char *str = "hello world";
+    utable_add_integer_array(table, "key1", int_array, 2);
+    utable_add_cstring(table, "key2", str, strlen(str));
+    utable_add_integer_array(table, "key3", int_array, 2);
+
+    EXPECT_STREQ(utable_next_key(table), "key1");
+    EXPECT_STREQ(utable_next_key(table), "key2");
+    EXPECT_STREQ(utable_next_key(table), "key3");
+    EXPECT_EQ(utable_next_key(table), nullptr);
+    utable_reset_iter(table);
+    EXPECT_STREQ(utable_next_key(table), "key1");
+    EXPECT_STREQ(utable_next_key(table), "key2");
+    EXPECT_STREQ(utable_next_key(table), "key3");
+    EXPECT_EQ(utable_next_key(table), nullptr);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=3;
+    B.n_item_size=strlen("key1")+sizeof(int64_t)*2+strlen("key2")+strlen(str)+strlen("key3")+sizeof(int64_t)*2;
+    B.n_blob=0;
+    B.n_blob_size=0;
+    B.n_cstring=1;
+    B.n_cstring_size=strlen(str);
+    B.n_integer_array=2;
+    B.n_integer_array_size=sizeof(int64_t)*4;
+    B.n_integer=0;
+    B.n_integer_size=0;
+    utable_stat(table, &A);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+
+    utable_free(table);
+}
+
+TEST(utable_test, dup_empty)
+{
+    struct utable *table = utable_new();
+
+    struct utable *dup_table = utable_duplicate(table);
+    EXPECT_EQ(utable_next_key(dup_table), nullptr);
+
+    utable_free(table);
+    utable_free(dup_table);
+}
+
+TEST(utable_test, dup_and_query)
+{
+    struct utable *table = utable_new();
+    
+    const char *str = "hello world";
+    utable_add_cstring(table, "key1", str, strlen(str));
+    int64_t int_array[] = { 1, 2 };
+    utable_add_integer_array(table, "key2", int_array, 2);
+    const char *blob = "bin hello world";
+    utable_add_blob(table, "key3", blob, strlen(blob));
+
+    struct utable *dup_table = utable_duplicate(table);
+    test_utable_assert_arr(dup_table, "key2", int_array, 2);
+    test_utable_assert_str(dup_table, "key1", str);
+    test_utable_assert_blob(dup_table, "key3", blob, strlen(blob));
+
+    struct utable_stat A={}, B={};
+    utable_stat(table, &A);
+    utable_stat(dup_table, &B);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+    utable_free(dup_table);
+}
+
+TEST(utable_test, replace_8k_cstring)
+{
+    struct utable *table = utable_new();
+    
+    const char str1[] = "hello world";
+    char str2[8192];
+    memset(str2, 'B', sizeof(str2));
+    str2[8191] = '\0'; // make sure it's null-terminated
+    
+    utable_add_cstring(table, "key1", str1, strlen(str1));
+
+    test_utable_assert_str(table, "key1", str1);
+    
+    utable_delete_item(table, "key1");
+    utable_add_cstring(table, "key1", str2, strlen(str2));
+
+    test_utable_assert_str(table, "key1", str2);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key1")+strlen(str2);
+    B.n_cstring=1;
+    B.n_cstring_size=strlen(str2);
+    utable_stat(table, &A);
+
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+}
+
+TEST(utable_test, replace_cstring_many_times)
+{
+    struct utable *table = utable_new();
+    
+    char str[4000] ;
+    
+    memset(str, 'B', sizeof(str));
+    str[3999] = '\0'; // make sure it's null-terminated
+
+    utable_add_cstring(table, "key1", str, strlen(str));
+
+    test_utable_assert_str(table, "key1", str);
+    
+    for(int i=0; i<100000; i++) 
+    {
+        utable_delete_item(table, "key1");
+        utable_add_cstring(table, "key1", str, strlen(str));
+    }
+
+    test_utable_assert_str(table, "key1", str);
+
+    struct utable_stat A={};
+    struct utable_stat B={};
+    B.n_item=1;
+    B.n_item_size=strlen("key1")+strlen(str);
+    B.n_cstring=1;
+    B.n_cstring_size=strlen(str);
+    utable_stat(table, &A);
+
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+    printf("replace_cstring_many_times done\n");
+    utable_free(table);
+}
+
+TEST(utable_test, merge_empty_to_empty)
+{
+    struct utable *table = utable_new();
+    struct utable *table2 = utable_new();
+    utable_merge(table, table2);
+    EXPECT_EQ(utable_next_key(table), nullptr);
+
+    struct utable_stat A={}, B={};
+    utable_stat(table, &A);
+    utable_stat(table2, &B);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+    utable_free(table2);
+}
+
+TEST(utable_test, merge_all_new)
+{
+    struct utable *table = utable_new();
+    struct utable *table2 = utable_new();
+    
+    const char *str = "hello world";
+    utable_add_cstring(table, "key1", str, strlen(str));
+    int64_t int_array[] = { 1, 2 };
+    utable_add_integer_array(table, "key2", int_array, 2);
+
+    const char *blob = "bin hello world";
+    utable_add_blob(table2, "key on tbl2", blob, strlen(blob));
+
+    utable_merge(table2, table);
+    test_utable_assert_arr(table2, "key2", int_array, 2);
+    test_utable_assert_str(table2, "key1", str);
+    test_utable_assert_blob(table2, "key on tbl2", blob, strlen(blob));
+
+    utable_merge(table, table2);
+    struct utable_stat A={}, B={};
+    utable_stat(table, &A);
+    utable_stat(table2, &B);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+    utable_free(table2);
+}
+
+TEST(utable_test, merge_all_skip)
+{
+    struct utable *table = utable_new();
+    const char *str = "hello world";
+    utable_add_cstring(table, "key1", str, strlen(str));
+    int64_t int_array[] = { 1, 2 };
+    utable_add_integer_array(table, "key2", int_array, 2);
+    struct utable *table2 = utable_duplicate(table);
+
+    utable_merge(table2, table);
+
+    EXPECT_STREQ(utable_next_key(table2), "key1");
+    EXPECT_STREQ(utable_next_key(table2), "key2");
+    EXPECT_EQ(utable_next_key(table2), nullptr);
+
+    utable_free(table);
+    utable_free(table2);
+}
+
+TEST(utable_test, merge_some_skip_some_new)
+{
+    struct utable *table = utable_new();    
+    struct utable *table2 = utable_new();
+
+    const char *str = "val on tbl1";
+    utable_add_cstring(table, "key_share", str, strlen(str));
+    const char *str2 = "val on tbl2";
+    utable_add_cstring(table2, "key_share", str2, strlen(str2));
+    const char *str3 = "val on tbl1";
+    utable_add_cstring(table, "key1", str3, strlen(str3));
+
+    utable_merge(table2, table);
+
+    test_utable_assert_str(table2, "key_share", str2);
+    test_utable_assert_str(table2, "key1", str3);
+
+    utable_free(table);
+    utable_free(table2);
+}
+
+TEST(utable_test, serialize_and_deserialize)
+{
+    struct utable *table = utable_new();
+    int64_t int_array[] = { 1, 2 };
+    const char *str = "hello world";
+    const char *blob = "bin hello world";
+    utable_add_integer_array(table, "key1", int_array, 2);
+    utable_add_cstring(table, "key2", str, strlen(str));
+    utable_add_blob(table, "key3", blob, strlen(blob));
+
+    char *blob_out;
+    size_t blob_size;
+    utable_serialize(table, &blob_out, &blob_size);
+
+    struct utable *table2 = utable_deserialize(blob_out, blob_size);
+    test_utable_assert_arr(table2, "key1", int_array, 2);
+    test_utable_assert_str(table2, "key2", str);
+    test_utable_assert_blob(table2, "key3", blob, strlen(blob));
+
+    struct utable_stat A={}, B={};
+    utable_stat(table, &A);
+    utable_stat(table2, &B);
+    EXPECT_EQ(memcmp(&A, &B, sizeof(struct utable_stat)), 0);
+
+    utable_free(table);
+    free(blob_out);
+    utable_free(table2);
+}
+
+TEST(utable_test, serialize_empty_and_deserialize)
+{
+    struct utable *table = utable_new();
+    
+    char *blob_out;
+    size_t blob_size;
+    utable_serialize(table, &blob_out, &blob_size);
+
+    struct utable *table2 = utable_deserialize(blob_out, blob_size);
+
+    ASSERT_TRUE(table2 != NULL);
+    ASSERT_EQ(utable_next_key(table2), nullptr);
+
+    utable_free(table);
+    free(blob_out);
+    utable_free(table2);
+}
+
+int main(int argc, char *argv[]) 
+{
+	testing::InitGoogleTest(&argc, argv);
+	return RUN_ALL_TESTS();
+}

deps/utable/utable.c

@@ -0,0 +1,730 @@
+#include "utable.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "mpack/mpack.h"
+
+#define YYJSON_HAS_STDINT_H 1
+#define YYJSON_DISABLE_READER 1
+#define YYJSON_HAS_STDBOOL_H 1
+#include "yyjson/yyjson.h"
+
+#include "uthash/uthash.h"
+#include "base64/b64.h"
+
+#include "nmx_pool/mempool.h"
+
+#ifdef DEBUG_MODE
+    #define DEBUG_PRINT(...) printf(__VA_ARGS__)
+#else
+    #define DEBUG_PRINT(...) 
+#endif
+
+#define ALLOC(number, type) ((type *)calloc(sizeof(type), number))
+#define	FREE(p)	{free(p);p=NULL;}
+
+#define MEMPOOL_TYPE mem_pool_t
+
+#define MEMPOOL_CREATE(pool_size) create_mem_pool(pool_size)
+#define MEMPOOL_DESTROY(pool) destroy_mem_pool(pool)
+
+#define MEMPOOL_ALLOC(pool, number, type) ((type *)mem_alloc(pool, sizeof(type) * number))
+#define MEMPOOL_FREE(pool, p) mem_free(pool, p)
+
+
+
+struct utable_item {
+    char *key;
+    size_t key_sz;
+    enum utable_value_type value_type;
+    size_t value_sz;
+    union {
+        struct 
+        {
+            char *cstring;
+            size_t cstring_sz;
+        };
+        struct
+        {
+            char *blob;
+            size_t blob_sz;
+        };
+        struct
+        {
+            int64_t *interger_array;
+            size_t n_integer;
+        };
+        struct
+        {
+            char **cstring_array;
+            size_t *cstring_array_sz;
+            size_t n_cstring;
+        };        
+        int64_t integer;
+    };
+    UT_hash_handle hh;
+};
+
+struct utable {
+    struct utable_item *items;
+    struct utable_item *iter;
+    MEMPOOL_TYPE *mempool;
+    struct utable_stat stat;
+};
+
+
+
+
+struct utable *utable_new_with_size(size_t sz)
+{
+    struct utable *table = ALLOC(1, struct utable);
+    table->mempool = MEMPOOL_CREATE(sz);
+    return table; 
+}
+
+#define UTABLE_INITIAL_MEMPOOL_SIZE 2 * 1024
+
+inline struct utable *utable_new(void)
+{
+    return utable_new_with_size(UTABLE_INITIAL_MEMPOOL_SIZE);
+}
+
+void utable_free(struct utable *table)
+{
+    if(table->items) HASH_CLEAR(hh, table->items);
+    if(table->mempool) MEMPOOL_DESTROY(table->mempool);
+    FREE(table);
+}
+
+inline void utable_stat(struct utable *table, struct utable_stat *stat)
+{
+    if(stat == NULL)
+        return;
+    memcpy(stat, &table->stat, sizeof(struct utable_stat));
+}
+
+static void utable_item_stat_add(struct utable_stat *stat, struct utable_item *item)
+{
+    if(stat==NULL || item == NULL)
+        return;
+    stat->n_item++;
+    stat->n_item_size += item->key_sz;
+    stat->n_item_size += item->value_sz;
+    switch (item->value_type) 
+    {
+        case utable_value_type_cstring:
+            stat->n_cstring++;
+            stat->n_cstring_size += item->value_sz;
+            break;
+        case utable_value_type_blob:
+            stat->n_blob++;
+            stat->n_blob_size += item->value_sz;
+            break;
+        case utable_value_type_integer:
+            stat->n_integer++;
+            stat->n_integer_size += item->value_sz;
+            break;
+        case utable_value_type_integer_array:
+            stat->n_integer_array++;
+            stat->n_integer_array_size += item->value_sz;
+            break;
+        case utable_value_type_cstring_array:
+            stat->n_cstring_array++;
+            stat->n_cstring_array_size += item->value_sz;
+            break;            
+        default:
+            break;
+    }
+}
+
+static void utable_item_stat_sub(struct utable_stat *stat, struct utable_item *item)
+{
+    if(stat==NULL || item == NULL)
+        return;
+    stat->n_item--;
+    stat->n_item_size -= item->key_sz;
+    stat->n_item_size -= item->value_sz;
+    switch (item->value_type) 
+    {
+        case utable_value_type_cstring:
+            stat->n_cstring--;
+            stat->n_cstring_size -= item->cstring_sz;
+            break;
+        case utable_value_type_blob:
+            stat->n_blob--;
+            stat->n_blob_size -= item->blob_sz;
+            break;
+        case utable_value_type_integer:
+            stat->n_integer--;
+            stat->n_integer_size -= sizeof(item->integer);
+            break;
+        case utable_value_type_integer_array:
+            stat->n_integer_array--;
+            stat->n_integer_array_size -= sizeof(int64_t) * item->n_integer;
+            break;
+        case utable_value_type_cstring_array:
+            stat->n_cstring_array--;
+            stat->n_cstring_array_size -= item->value_sz;
+            break;               
+        default:
+            break;
+    }
+}
+
+static char*mempool_strdup(MEMPOOL_TYPE  *mempool, const char* s, size_t len) 
+{
+    char* new_str = MEMPOOL_ALLOC(mempool, len + 1, char);  
+    memcpy(new_str, s, len + 1);
+    return new_str;  
+}
+
+void utable_add_cstring(struct utable *table, const char *key, const char *value, size_t value_sz)
+{
+    // check if key already exists
+    struct utable_item *item;
+    size_t key_sz = strlen(key);
+    HASH_FIND(hh, table->items, key, key_sz, item);
+    if (item) {
+        DEBUG_PRINT("ERR: key %s already exists\n", key);
+        return;
+    }
+    item = MEMPOOL_ALLOC(table->mempool ,1,struct utable_item);
+    item->key = mempool_strdup(table->mempool,key, key_sz);
+    item->key_sz = key_sz;
+    item->value_type = utable_value_type_cstring;
+    item->cstring = MEMPOOL_ALLOC(table->mempool ,value_sz + 1, char);
+    memcpy(item->cstring, value, value_sz);
+    item->cstring[value_sz] = '\0';
+    item->cstring_sz = value_sz;
+    item->value_sz = value_sz;
+    HASH_ADD_KEYPTR(hh, table->items, item->key, item->key_sz, item);
+    utable_item_stat_add(&table->stat, item);
+}
+
+void utable_add_blob(struct utable *table, const char *key, const char *blob, size_t blob_sz)
+{
+    // check if key already exists
+    struct utable_item *item;
+    size_t key_sz = strlen(key);
+    HASH_FIND(hh, table->items, key, key_sz, item);
+    if (item) {
+        DEBUG_PRINT("ERR: key %s already exists\n", key);
+        return;
+    }
+    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
+    item->key = mempool_strdup(table->mempool, key, key_sz);
+    item->key_sz = key_sz;
+    item->value_type = utable_value_type_blob;
+    item->blob = MEMPOOL_ALLOC(table->mempool ,blob_sz, char);
+    memcpy(item->blob, blob, blob_sz);
+    item->blob_sz = blob_sz;
+    item->value_sz = blob_sz;
+    HASH_ADD_KEYPTR(hh, table->items, item->key, item->key_sz, item);
+
+    utable_item_stat_add(&table->stat, item);
+}
+
+void utable_add_integer(struct utable *table, const char *key, int64_t value)
+{
+    // check if key already exists
+    struct utable_item *item;
+    size_t key_sz = strlen(key);
+    HASH_FIND(hh, table->items, key, key_sz, item);
+    if (item) {
+        DEBUG_PRINT("ERR: key %s already exists\n", key);
+        return;
+    }
+    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
+    item->key = mempool_strdup(table->mempool, key, key_sz);
+    item->key_sz = key_sz;
+    item->value_type = utable_value_type_integer;
+    item->integer = value;
+    item->value_sz = sizeof(int64_t);
+    HASH_ADD_KEYPTR(hh, table->items, item->key, item->key_sz, item);
+
+    utable_item_stat_add(&table->stat, item); 
+}
+
+void utable_add_integer_array(struct utable *table, const char *key, int64_t value_array[], size_t n_value)
+{
+    // check if key already exists
+    struct utable_item *item;
+    size_t key_sz = strlen(key);
+    HASH_FIND(hh, table->items, key, key_sz, item);
+    if (item) {
+        DEBUG_PRINT("ERR: key %s already exists\n", key);
+        return;
+    }
+    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
+    item->key = mempool_strdup(table->mempool, key, key_sz);
+    item->key_sz = key_sz;
+    item->value_type = utable_value_type_integer_array;
+    item->interger_array = MEMPOOL_ALLOC(table->mempool ,n_value, int64_t);
+    memcpy(item->interger_array, value_array, sizeof(int64_t) * n_value);
+    item->n_integer = n_value;
+    item->value_sz = sizeof(int64_t) * n_value;
+    HASH_ADD_KEYPTR(hh, table->items, item->key, item->key_sz, item);
+
+    utable_item_stat_add(&table->stat, item);    
+}
+
+void utable_add_cstring_array(struct utable *table, const char *key, const char* value_array[], size_t value_sz[], size_t n_value)
+{
+    // check if key already exists
+    struct utable_item *item;
+    size_t key_sz = strlen(key);
+    HASH_FIND(hh, table->items, key, key_sz, item);
+    if (item) {
+        DEBUG_PRINT("ERR: key %s already exists\n", key);
+        return;
+    }
+    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
+    item->key = mempool_strdup(table->mempool, key, key_sz);
+    item->key_sz = key_sz;
+    item->value_type = utable_value_type_cstring_array;
+    item->cstring_array = MEMPOOL_ALLOC(table->mempool ,n_value, char *);
+    item->cstring_array_sz = MEMPOOL_ALLOC(table->mempool ,n_value, size_t);
+    item->value_sz = 0;
+    for(size_t i =0; i < n_value; i++)
+    {
+        item->cstring_array[i] = MEMPOOL_ALLOC(table->mempool , value_sz[i]+1, char);
+        memcpy(item->cstring_array[i], value_array[i], value_sz[i]);
+        item->cstring_array[i][value_sz[i]] = '\0';
+        item->cstring_array_sz[i] = value_sz[i];
+        item->value_sz += value_sz[i];
+    }
+    item->n_cstring = n_value;
+    HASH_ADD_KEYPTR(hh, table->items, item->key, item->key_sz, item);
+
+    utable_item_stat_add(&table->stat, item);    
+}
+
+void utable_delete_item(struct utable *table, const char *key)
+{
+    struct utable_item *item;
+    HASH_FIND_STR(table->items, key, item);
+    if (item) {
+        HASH_DEL(table->items, item);
+        MEMPOOL_FREE(table->mempool, item->key);
+        switch (item->value_type) {
+            case utable_value_type_cstring:
+                MEMPOOL_FREE(table->mempool,item->cstring);
+                break;
+            case utable_value_type_blob:
+                MEMPOOL_FREE(table->mempool,item->blob);
+                break;
+            case utable_value_type_integer_array:
+                MEMPOOL_FREE(table->mempool,item->interger_array);
+                break;
+            case utable_value_type_cstring_array:
+                for(size_t i=0; i < item->n_cstring; i++)
+                {
+                    MEMPOOL_FREE(table->mempool,item->cstring_array[i]);
+                }
+                MEMPOOL_FREE(table->mempool,item->cstring_array_sz);
+                break;                
+            default:
+                break;
+        }
+        utable_item_stat_sub(&table->stat, item);
+        MEMPOOL_FREE(table->mempool,item);
+    }
+}
+
+void utable_reset_iter(struct utable *table)
+{
+    table->iter = NULL;
+}
+
+const char *utable_next_key(struct utable *table)
+{
+    if (table->iter == NULL) {
+        table->iter = table->items;
+    } else {
+        table->iter = (struct utable_item *)table->iter->hh.next;
+    }
+    if (table->iter == NULL) {
+        return NULL;
+    }
+    return table->iter->key;
+}
+
+enum utable_value_type utable_get_value_type(const struct utable *table, const char *key)
+{
+    struct utable_item *item;
+    HASH_FIND_STR(table->items, key, item);
+    if (item) {
+        return item->value_type;
+    }
+    return utable_value_type_undefined;
+}
+
+int utable_get0_blob_value(const struct utable *table, const char *key, char **value, size_t *value_len)
+{
+    struct utable_item *item;
+    HASH_FIND_STR(table->items, key, item);
+    if (item) {
+        if (item->value_type == utable_value_type_blob) {
+            *value = item->blob;
+            *value_len = item->blob_sz;
+            return 0;
+        }
+    }
+    return -1;
+}
+
+int utable_get0_cstring_value(const struct utable *table, const char *key, char **value, size_t *value_len)
+{
+    struct utable_item *item;
+    HASH_FIND_STR(table->items, key, item);
+    if (item) {
+        if (item->value_type == utable_value_type_cstring) {
+            *value = item->cstring;
+            *value_len = item->cstring_sz;
+            return 0;
+        }
+    }
+    return -1;
+}
+
+int utable_get0_integer_value(const struct utable *table, const char *key, int64_t *value)
+{
+    struct utable_item *item;
+    HASH_FIND_STR(table->items, key, item);
+    if (item) {
+        if (item->value_type == utable_value_type_integer) {
+            *value = item->integer;
+            return 0;
+        }
+    }
+    return -1;
+}
+
+int utable_get0_integer_value_array(const struct utable *table, const char *key, int64_t **value_array, size_t *n_value)
+{
+    struct utable_item *item;
+    HASH_FIND_STR(table->items, key, item);
+    if (item) {
+        if (item->value_type == utable_value_type_integer_array) {
+            *value_array = item->interger_array;
+            *n_value = item->n_integer;
+            return 0;
+        }
+    }
+    return -1;
+}
+
+int utable_get0_cstring_value_array(const struct utable *table, const char *key, char ***value_array, size_t **value_len, size_t *n_value)
+{
+    struct utable_item *item;
+    HASH_FIND_STR(table->items, key, item);
+    if (item) {
+        if (item->value_type == utable_value_type_cstring_array) {
+            *value_array = item->cstring_array;
+            *value_len = item->cstring_array_sz;
+            *n_value = item->n_cstring;
+            return 0;
+        }
+    }
+    return -1;
+}
+
+struct utable_item *utable_item_dup(MEMPOOL_TYPE *mempool, const struct utable_item *item)
+{
+    struct utable_item *new_item = MEMPOOL_ALLOC(mempool, 1, struct utable_item);
+    size_t key_sz = item->key_sz;
+    new_item->key = mempool_strdup(mempool, item->key, key_sz);
+    new_item->key_sz = key_sz;
+    new_item->value_type = item->value_type;
+    new_item->value_sz= item->value_sz;
+    switch (item->value_type) {
+        case utable_value_type_cstring:
+            new_item->cstring = MEMPOOL_ALLOC(mempool,item->cstring_sz + 1, char);
+            memcpy(new_item->cstring, item->cstring, item->cstring_sz);
+            new_item->cstring[item->cstring_sz] = '\0';
+            new_item->cstring_sz = item->cstring_sz;
+            break;
+        case utable_value_type_blob:
+            new_item->blob = MEMPOOL_ALLOC(mempool,item->blob_sz, char);
+            memcpy(new_item->blob, item->blob, item->blob_sz);
+            new_item->blob_sz = item->blob_sz;
+            break;
+        case utable_value_type_integer:
+            new_item->integer = item->integer;
+            break;
+        case utable_value_type_integer_array:
+            new_item->interger_array = MEMPOOL_ALLOC(mempool,item->n_integer, int64_t);
+            memcpy(new_item->interger_array, item->interger_array, sizeof(int64_t) * item->n_integer);
+            new_item->n_integer = item->n_integer;
+            break;
+        case utable_value_type_cstring_array:
+            new_item->n_cstring= item->n_cstring;
+            new_item->cstring_array = MEMPOOL_ALLOC(mempool ,item->n_cstring, char *);
+            new_item->cstring_array_sz = MEMPOOL_ALLOC(mempool ,item->n_cstring, size_t);
+            for(size_t i =0; i < item->n_cstring; i++)
+            {
+                new_item->cstring_array[i] = MEMPOOL_ALLOC(mempool , item->cstring_array_sz[i]+1, char);
+                memcpy(new_item->cstring_array[i], item->cstring_array[i], item->cstring_array_sz[i]);
+                new_item->cstring_array[i][item->cstring_array_sz[i]] = '\0';
+                new_item->cstring_array_sz[i] = item->cstring_array_sz[i];
+            }
+            break;            
+        default:
+            break;
+    }
+    return new_item;
+}
+
+struct utable *utable_duplicate(const struct utable *table)
+{
+    struct utable *new_table = utable_new();
+    struct utable_item *item, *tmp;
+    HASH_ITER(hh, table->items, item, tmp) {
+        struct utable_item *new_item = utable_item_dup(new_table->mempool, item);
+        HASH_ADD_KEYPTR(hh, new_table->items, new_item->key, new_item->key_sz, new_item);
+    }
+    new_table->stat = table->stat;
+    return new_table;
+}
+
+int utable_merge(struct utable *dst, const struct utable *src)
+{
+    struct utable_item *item, *tmp;
+    HASH_ITER(hh, src->items, item, tmp) {
+        struct utable_item *dst_item;
+        HASH_FIND(hh, dst->items, item->key, item->key_sz, dst_item);
+        if (dst_item) {
+            continue;
+        }
+        struct utable_item *new_item = utable_item_dup(dst->mempool, item);
+        HASH_ADD_KEYPTR(hh, dst->items, new_item->key, new_item->key_sz, new_item);
+        utable_item_stat_add(&dst->stat, item);
+    }
+    return 0;
+}
+
+int utable_json_export(const struct utable *table, char **blob, size_t *blob_len)
+{
+    yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL);
+    yyjson_mut_val *root = yyjson_mut_obj(doc);
+    yyjson_mut_doc_set_root(doc, root);
+    char *encs[HASH_COUNT(table->items)];
+    struct utable_item *item, *tmp;
+    int enc_cnt = 0;
+    HASH_ITER(hh, table->items, item, tmp) {
+        switch (item->value_type) {
+            case utable_value_type_cstring: {
+                    yyjson_mut_obj_add_str(doc, root, item->key, item->cstring); // key and cstring are shallow copied
+                }
+                break;
+            case utable_value_type_blob: {
+                    char *enc = b64_encode((unsigned char *)item->blob, item->blob_sz);
+                    yyjson_mut_obj_add_str(doc, root, item->key, enc);
+                    // do not free enc now, it is shallow copied
+                    encs[enc_cnt++] = enc;
+                }
+                break;
+            case utable_value_type_integer: {
+                    yyjson_mut_obj_add_int(doc, root, item->key, item->integer);
+                }
+                break;
+            case utable_value_type_integer_array: {
+                    yyjson_mut_val *arr = yyjson_mut_arr_with_sint64(doc, item->interger_array, item->n_integer);
+                    yyjson_mut_obj_add_val(doc, root, item->key, arr);
+            }
+                break;
+            case utable_value_type_cstring_array: {
+                    yyjson_mut_val *arr = yyjson_mut_arr_with_strn(doc, (const char **)item->cstring_array, item->cstring_array_sz, item->n_cstring);
+                    yyjson_mut_obj_add_val(doc, root, item->key, arr);
+            }
+                break;                
+            default:
+                break;
+        }
+    }
+
+    char *json_str = yyjson_mut_write(doc, YYJSON_WRITE_ALLOW_INVALID_UNICODE, blob_len);
+    yyjson_mut_doc_free(doc);
+    *blob = json_str;
+    for (int j = 0; j < enc_cnt; j++) {
+        free(encs[j]);
+    }
+    return 0;
+}
+
+/*
+    [{
+        "key":<tag key string>
+        "type":"i_arr", "c_str", or "blob"
+        "value":<
+    }]
+*/
+int utable_msgpack_export(const struct utable *table, char **blob, size_t *blob_len)
+{
+    mpack_writer_t writer;
+    mpack_writer_init_growable(&writer, blob, blob_len);
+    mpack_start_array(&writer, HASH_COUNT(table->items));
+
+    struct utable_item *item, *tmp;
+    HASH_ITER(hh, table->items, item, tmp) {
+        mpack_start_map(&writer, 3);
+        mpack_write_cstr(&writer, "key");
+        mpack_write_cstr(&writer, item->key);
+
+        mpack_write_cstr(&writer, "type");
+        switch (item->value_type) {
+            case utable_value_type_cstring:
+                mpack_write_cstr(&writer, "c_str");
+                break;
+            case utable_value_type_blob:
+                mpack_write_cstr(&writer, "blob");
+                break;
+            case utable_value_type_integer:
+                mpack_write_cstr(&writer, "i");
+                break;
+            case utable_value_type_integer_array:
+                mpack_write_cstr(&writer, "i_arr");
+                break;
+            default:
+                break;
+        }
+
+        mpack_write_cstr(&writer, "value");
+        switch (item->value_type) {
+            case utable_value_type_cstring:
+                mpack_write_bin(&writer, item->cstring, item->cstring_sz);
+                break;
+            case utable_value_type_blob:
+                mpack_write_bin(&writer, item->blob, item->blob_sz);
+                break;
+            case utable_value_type_integer:
+                mpack_write_i64(&writer, item->integer);
+                break;
+            case utable_value_type_integer_array:
+                mpack_start_array(&writer, item->n_integer);
+                for (size_t i = 0; i < item->n_integer; i++) {
+                    mpack_write_i64(&writer, item->interger_array[i]);
+                }
+                mpack_finish_array(&writer);
+                break;
+            default:
+                break;
+        }
+
+        mpack_finish_map(&writer);
+    }
+
+    mpack_finish_array(&writer);
+    if (mpack_writer_destroy(&writer) != mpack_ok) {
+        DEBUG_PRINT("ERR mpack writer fieldtag_list_serialize destroy failed\n");
+        return -1;
+    }
+    return 0;
+}
+
+void utable_serialize(const struct utable *table, char **blob, size_t *blob_len) {
+    utable_msgpack_export(table, blob, blob_len);
+}
+
+char *get_cstring_from_mpack_node(MEMPOOL_TYPE *mempool ,mpack_node_t node)
+{
+    size_t len = mpack_node_strlen(node);
+    char *str = MEMPOOL_ALLOC(mempool, len + 1, char);
+    memcpy(str, mpack_node_str(node), len);
+    str[len] = '\0';
+    return str;
+}
+
+static void utable_item_free(MEMPOOL_TYPE *mempool, struct utable_item *item)
+{
+    if(item == NULL)
+        return;
+    if(item->key)MEMPOOL_FREE(mempool, item->key);
+    switch (item->value_type) {
+        case utable_value_type_cstring:
+            MEMPOOL_FREE(mempool,item->cstring);
+            break;
+        case utable_value_type_blob:
+            MEMPOOL_FREE(mempool,item->blob);
+            break;
+        case utable_value_type_integer_array:
+            MEMPOOL_FREE(mempool,item->interger_array);
+            break;
+        case utable_value_type_cstring_array:
+            for(size_t i=0; i < item->n_cstring; i++)
+            {
+                MEMPOOL_FREE(mempool,item->cstring_array[i]);
+            }
+            MEMPOOL_FREE(mempool,item->cstring_array_sz);
+            break;            
+        default:
+            break;
+    }
+}
+
+struct utable *utable_deserialize(const char *blob, size_t blob_len)
+{
+    mpack_tree_t tree;
+    mpack_tree_init_data(&tree, blob, blob_len);
+    mpack_tree_parse(&tree);
+    mpack_node_t root = mpack_tree_root(&tree);
+    if (mpack_node_type(root) != mpack_type_array) {
+        DEBUG_PRINT("ERR mpack root type is not array\n");
+        return NULL;
+    }
+    size_t n_item = mpack_node_array_length(root);
+
+    struct utable *table = utable_new();
+    for (size_t i = 0; i < n_item; i++) {
+        mpack_node_t item = mpack_node_array_at(root, i);
+        mpack_node_t key = mpack_node_map_cstr(item, "key");
+        mpack_node_t type = mpack_node_map_cstr(item, "type");
+        mpack_node_t value = mpack_node_map_cstr(item, "value");
+        char *type_str = get_cstring_from_mpack_node(table->mempool, type);
+
+        struct utable_item *new_item = MEMPOOL_ALLOC(table->mempool,1, struct utable_item);
+        new_item->key = get_cstring_from_mpack_node(table->mempool, key);
+        new_item->key_sz = strlen(new_item->key);
+        new_item->value_sz = 0;
+    
+        if (strcmp(type_str, "c_str") == 0) {
+            new_item->value_type = utable_value_type_cstring;
+            new_item->cstring = MEMPOOL_ALLOC(table->mempool,mpack_node_bin_size(value), char);
+            new_item->cstring_sz = mpack_node_bin_size(value);
+            new_item->value_sz = new_item->cstring_sz;
+            memcpy(new_item->cstring, mpack_node_bin_data(value), mpack_node_bin_size(value));
+        } else if (strcmp(type_str, "blob") == 0) {
+            new_item->value_type = utable_value_type_blob;
+            new_item->blob = MEMPOOL_ALLOC(table->mempool,mpack_node_bin_size(value), char);
+            new_item->blob_sz = mpack_node_bin_size(value);
+            new_item->value_sz = new_item->blob_sz;
+            memcpy(new_item->blob, mpack_node_bin_data(value), mpack_node_bin_size(value));
+        } else if (strcmp(type_str, "i") == 0) {
+            new_item->value_type = utable_value_type_integer;
+            new_item->integer = mpack_node_i64(value);
+        } else if (strcmp(type_str, "i_arr") == 0) {
+            new_item->value_type = utable_value_type_integer_array;
+            new_item->n_integer = mpack_node_array_length(value);
+            new_item->interger_array = MEMPOOL_ALLOC(table->mempool,new_item->n_integer, int64_t);
+            for (size_t j = 0; j < new_item->n_integer; j++) {
+                new_item->interger_array[j] = mpack_node_i64(mpack_node_array_at(value, j));
+            }
+            new_item->value_sz = sizeof(int64_t) * new_item->n_integer;
+        } else {
+            DEBUG_PRINT("ERR unknown type %s\n", type_str);
+            mpack_tree_destroy(&tree);
+            MEMPOOL_FREE(table->mempool, type_str);
+            utable_item_free(table->mempool, new_item);
+            return NULL;
+        }
+
+        HASH_ADD_KEYPTR(hh, table->items, new_item->key, new_item->key_sz, new_item);
+        utable_item_stat_add(&table->stat, new_item);
+        MEMPOOL_FREE(table->mempool, type_str);
+    }
+
+    mpack_tree_destroy(&tree);
+    return table;
+}

deps/utable/utable.h

@@ -0,0 +1,87 @@
+#pragma once
+
+#include <stdint.h>
+#include <stddef.h>
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+struct utable_stat
+{
+    size_t n_item;
+    size_t n_item_size;
+    size_t n_blob;
+    size_t n_blob_size;
+    size_t n_cstring;
+    size_t n_cstring_size;
+    size_t n_integer;
+    size_t n_integer_size;
+    size_t n_integer_array;
+    size_t n_integer_array_size;
+    size_t n_cstring_array;
+    size_t n_cstring_array_size;
+};
+
+struct utable;
+struct utable *utable_new(void);
+struct utable *utable_new_with_size(size_t sz);
+void utable_free(struct utable *table);
+
+
+
+
+
+void utable_stat(struct utable *table, struct utable_stat *stat);
+
+void utable_add_cstring(struct utable *table, const char *key, const char *value, size_t value_sz);
+void utable_add_blob(struct utable *table, const char *key, const char *blob, size_t blob_sz);
+void utable_add_integer(struct utable *table, const char *key, int64_t value);
+void utable_add_integer_array(struct utable *table, const char *key, int64_t value_array[], size_t n_value);
+void utable_add_cstring_array(struct utable *table, const char *key, const char* value_array[], size_t value_sz[], size_t n_value);
+void utable_delete_item(struct utable *table, const char *key);
+
+enum utable_value_type
+{
+    utable_value_type_undefined=0,
+    utable_value_type_cstring,
+    utable_value_type_blob,
+    utable_value_type_integer,
+    utable_value_type_integer_array,
+    utable_value_type_cstring_array
+};
+
+void utable_reset_iter(struct utable *table);
+const char *utable_next_key(struct utable *table);
+enum utable_value_type utable_get_value_type(const struct utable *table, const char *key);
+int utable_get0_blob_value(const struct utable *table, const char *key, char **value, size_t *value_len);
+int utable_get0_cstring_value(const struct utable *table, const char *key, char **value, size_t *value_len);
+int utable_get0_integer_value(const struct utable *table, const char *key, int64_t *value);
+int utable_get0_integer_value_array(const struct utable *table, const char *key, int64_t **value_array, size_t *n_value);
+int utable_get0_cstring_value_array(const struct utable *table, const char *key, char ***value_array, size_t **value_len, size_t *n_value);
+
+struct utable *utable_duplicate(const struct utable *table);
+int utable_merge(struct utable *dst, const struct utable *src);
+
+void utable_serialize(const struct utable *table, char **blob, size_t *blob_len);
+struct utable *utable_deserialize(const char *blob, size_t blob_len);
+
+struct ipfix_exporter_schema;
+struct ipfix_exporter_schema *utable_ipfix_exporter_schema_new(const char *ipfix_schema_json_path, uint16_t domain_id, uint16_t n_worker);  // |domain_id|work_id|=source_id (16+16) domain_id 是进程唯一标识
+void utable_ipfix_exporter_schema_free(struct ipfix_exporter_schema *ipfix_schema);
+
+uint32_t utable_ipfix_template_flow_refresh_interval_s(struct ipfix_exporter_schema *ipfix_schema);
+const char *utable_ipfix_template_flow_get0(struct ipfix_exporter_schema *ipfix_schema, uint16_t worker_id, size_t *blob_len);
+
+// return template id, -1 if not found
+int utable_ipfix_template_get(struct ipfix_exporter_schema *ipfix_schema, const char *template_name);
+int utable_ipfix_data_flow_exporter(const struct utable *table, struct ipfix_exporter_schema *ipfix_schema, int template_id, uint16_t worker_id, char **blob, size_t *blob_len);
+
+
+int utable_json_export(const struct utable *table, char **blob, size_t *blob_len);
+int utable_msgpack_export(const struct utable *table, char **blob, size_t *blob_len);
+
+#ifdef __cplusplus
+}
+#endif

deps/utable/utable_ipfix_exporter.c

@@ -0,0 +1,795 @@
+#include <stdio.h>
+#include <limits.h>
+#include <stdint.h>
+#include <stddef.h>
+#include <net/if.h>
+#include <netinet/in.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <time.h>
+
+#include "utable.h"
+#include "cjson/cJSON.h"
+#include "yyjson/yyjson.h"
+#include "uthash/utarray.h"
+#include "uthash/uthash.h"
+
+#define IPFIX_DEFUALT_VERSION 10
+#define GEEDGE_NETWORKS_PEN_NUMBER 54450
+#define IPFIX_DEFUALT_PEN_NUMBER GEEDGE_NETWORKS_PEN_NUMBER
+#define IPFIX_NONE_PEN_NUMBER -1
+#define IPFIX_TEMPLATE_SET_ID 2
+#define IPFIX_BUFF_MAX_SIZE 2048
+#define IPFIX_ELEMENT_MAX_LEN 32
+
+#ifndef MIN
+#define MIN(a, b) (((a) < (b)) ? (a) : (b))
+#endif
+
+enum ipfix_type
+{
+    IPFIX_UNSIGNED8 = 0,
+    IPFIX_UNSIGNED16,
+    IPFIX_UNSIGNED32,
+    IPFIX_UNSIGNED64,
+    IPFIX_VARIABLE_STRING,
+    IPFIX_UNKNOWN
+};
+
+struct ipfix_element
+{
+    enum ipfix_type type;
+    uint16_t element_id;
+    uint16_t value_length; // if type is variable_string, value_length is 0
+    int PEN_number;
+    char name[IPFIX_ELEMENT_MAX_LEN];
+};
+
+struct ipfix_template
+{
+    uint16_t template_id;
+    char *name;
+    UT_array *elements_array; // utarray for current template elements
+};
+
+struct ipfix_worker_context
+{
+    int source_id;
+    int sequence;
+    size_t template_blob_length;
+    char *template_blob;
+};
+
+struct ipfix_exporter_schema
+{
+    uint16_t n_worker;
+    uint16_t version;
+    int domain_id;
+    int PEN_number;
+    uint32_t templates_refresh_interval_s;
+    UT_array *templates_array;                         // utarray for templates
+    struct ipfix_worker_context *worker_context_array; // utarray for worker_context
+};
+
+struct ipfix_message_head
+{
+    uint16_t version;
+    uint16_t length;
+    int exporttime;
+    int ipfix_message_sequence;
+    int domain_id;
+};
+#define IPFIX_MESSAGE_HEAD_LEN sizeof(struct ipfix_message_head)
+
+// from maat
+int load_file_to_memory(const char *file_name, unsigned char **pp_out, size_t *out_sz)
+{
+    int ret = 0;
+    FILE *fp = NULL;
+    struct stat fstat_buf;
+    size_t read_size = 0;
+
+    ret = stat(file_name, &fstat_buf);
+    if (ret != 0)
+    {
+        return -1;
+    }
+
+    fp = fopen(file_name, "r");
+    if (fp == NULL)
+    {
+        return -1;
+    }
+
+    *out_sz = fstat_buf.st_size;
+    *pp_out = (unsigned char *)calloc(*out_sz + 1, sizeof(unsigned char));
+    read_size = fread(*pp_out, 1, *out_sz, fp);
+    if (read_size != *out_sz)
+    {
+        free(*pp_out);
+        *pp_out = NULL;
+    }
+
+    fclose(fp);
+    fp = NULL;
+    return 0;
+}
+
+int ipfix_exporter_get_type(struct ipfix_element *p_element, char *element_type)
+{
+    if (strcmp(element_type, "string") == 0)
+    {
+        p_element->type = IPFIX_VARIABLE_STRING;
+        p_element->value_length = 0;
+        return 0;
+    }
+    else if (strcmp(element_type, "unsigned8") == 0)
+    {
+        p_element->type = IPFIX_UNSIGNED8;
+        p_element->value_length = sizeof(uint8_t);
+        return 0;
+    }
+    else if (strcmp(element_type, "unsigned16") == 0)
+    {
+        p_element->type = IPFIX_UNSIGNED16;
+        p_element->value_length = sizeof(uint16_t);
+        return 0;
+    }
+    else if (strcmp(element_type, "unsigned32") == 0)
+    {
+        p_element->type = IPFIX_UNSIGNED32;
+        p_element->value_length = sizeof(uint32_t);
+        return 0;
+    }
+    else if (strcmp(element_type, "unsigned64") == 0)
+    {
+        p_element->type = IPFIX_UNSIGNED64;
+        p_element->value_length = sizeof(uint64_t);
+        return 0;
+    }
+    else
+    {
+        return -1;
+    }
+}
+
+int ipfix_template_payload_init(UT_array *templates_array, char **blob, size_t *blob_len)
+{
+    if (templates_array == NULL || utarray_len(templates_array) == 0 || blob == NULL || blob_len == NULL)
+    {
+        return -1;
+    }
+
+    size_t n_templates = utarray_len(templates_array);
+    size_t template_blob_len = 0;
+    for (size_t i = 0; i < n_templates; i++)
+    {
+        struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(templates_array, i);
+        size_t n_elements = utarray_len(template_item->elements_array);
+        template_blob_len += (sizeof(uint16_t) + sizeof(uint16_t) + n_elements * (sizeof(uint16_t) + sizeof(uint16_t) + sizeof(uint32_t))); // template_id + field_count + element_id + length + PEN_number
+    }
+    template_blob_len += ((sizeof(uint16_t) + sizeof(uint16_t)) + IPFIX_MESSAGE_HEAD_LEN); // set_id + set_length + ipfix_message_head
+
+    char *payload = (char *)calloc(template_blob_len, sizeof(char));
+    int offset = 0;
+
+    offset += IPFIX_MESSAGE_HEAD_LEN;                                                    // skip ipfix_message_head
+    *(uint16_t *)(payload + offset) = htons(IPFIX_TEMPLATE_SET_ID);                      // set_id
+    offset += sizeof(uint16_t);                                                          // skip set_id
+    *(uint16_t *)(payload + offset) = htons(template_blob_len - IPFIX_MESSAGE_HEAD_LEN); // set_length
+    offset += sizeof(uint16_t);                                                          // skip set_length
+
+    for (size_t i = 0; i < n_templates; i++)
+    {
+        struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(templates_array, i);
+        size_t n_elements = utarray_len(template_item->elements_array);
+        *(uint16_t *)(payload + offset) = htons(template_item->template_id); // template_id
+        offset += sizeof(uint16_t);                                          // skip template_id
+        *(uint16_t *)(payload + offset) = htons(n_elements);                 // field_count
+        offset += sizeof(uint16_t);                                          // skip field_count
+
+        for (size_t j = 0; j < n_elements; j++)
+        {
+            struct ipfix_element *p_element = (struct ipfix_element *)utarray_eltptr(template_item->elements_array, j);
+
+            // set element_id
+            if (p_element->PEN_number != IPFIX_NONE_PEN_NUMBER)
+            {
+                *(uint16_t *)(payload + offset) = htons(p_element->element_id | 0x8000); // Pen_provided is yes
+            }
+            else
+            {
+                *(uint16_t *)(payload + offset) = htons(p_element->element_id); // Pen_provided is no
+            }
+            offset += sizeof(uint16_t); // skip element_id
+
+            // set element_length
+            if (p_element->type == IPFIX_VARIABLE_STRING)
+            {
+                *(uint16_t *)(payload + offset) = htons(65535); // if element_length is 65535, it means variable length
+            }
+            else
+            {
+                *(uint16_t *)(payload + offset) = htons(p_element->value_length); // length
+            }
+            offset += sizeof(uint16_t); // skip length
+
+            // set PEN_number
+            if (p_element->PEN_number != IPFIX_NONE_PEN_NUMBER)
+            {
+                *(uint32_t *)(payload + offset) = htonl(p_element->PEN_number); // PEN_number
+                offset += sizeof(uint32_t);                                     // skip PEN_number
+            }
+        }
+    }
+
+    if (offset != (int)template_blob_len)
+    {
+        free(payload);
+        payload = NULL;
+        return -1;
+    }
+
+    *blob = payload;
+    *blob_len = offset;
+    return 0;
+}
+
+void ipfix_template_item_destroy(void *elt)
+{
+    struct ipfix_template *template_item = (struct ipfix_template *)elt;
+    if (template_item == NULL)
+    {
+        return;
+    }
+
+    if (template_item->elements_array != NULL)
+    {
+        utarray_free(template_item->elements_array);
+    }
+
+    if (template_item->name != NULL)
+    {
+        free(template_item->name);
+        template_item->name = NULL;
+    }
+
+    return;
+}
+
+void ipfix_utarray_init(UT_array **_array, size_t sz_item, void (*dtor)(void *))
+{
+    if (_array == NULL || *_array != NULL)
+    {
+        return;
+    }
+
+    UT_icd icd = {0};
+    icd.sz = sz_item;
+    icd.dtor = dtor;
+    utarray_new(*_array, &icd);
+    return;
+}
+
+UT_array *ipfix_elements_array_init(cJSON *root, const char *element_key, int PEN_number_global)
+{
+    if (root == NULL || element_key == NULL)
+    {
+        return NULL;
+    }
+
+    cJSON *elements = cJSON_GetObjectItem(root, element_key);
+    if (elements == NULL)
+    {
+        return NULL;
+    }
+
+    UT_array *elements_array = NULL;
+    ipfix_utarray_init(&elements_array, sizeof(struct ipfix_element), NULL);
+    size_t n_elements = cJSON_GetArraySize(elements);
+    for (size_t i = 0; i < n_elements; i++)
+    {
+        struct ipfix_element element_item = {0};
+        cJSON *element = cJSON_GetArrayItem(elements, i);
+        cJSON *element_id = cJSON_GetObjectItem(element, "element_id");
+        if (element_id == NULL || element_id->valueint == 0) // element_id不存在、类型不为Number、值为0
+        {
+            goto error;
+        }
+        element_item.element_id = element_id->valueint;
+
+        cJSON *element_type = cJSON_GetObjectItem(element, "element_type");
+        if (element_type == NULL || element_type->valuestring == NULL) // element_type不存在、类型不为String、值为NULL
+        {
+            goto error;
+        }
+
+        if (ipfix_exporter_get_type(&element_item, element_type->valuestring) == -1) // element_type不在支持范围内
+        {
+            goto error;
+        }
+
+        cJSON *PEN_number = cJSON_GetObjectItem(element, "PEN_number");
+        if (PEN_number == NULL) // PEN_number不存在、类型不为Number
+        {
+            element_item.PEN_number = PEN_number_global;
+        }
+        else
+        {
+            element_item.PEN_number = PEN_number->valueint;
+        }
+
+        cJSON *element_name = cJSON_GetObjectItem(element, "element_name");
+        if (element_name == NULL || element_name->valuestring == NULL) // element_name不存在、类型不为String、值为NULL
+        {
+            goto error;
+        }
+        memcpy(element_item.name, element_name->valuestring, MIN(strlen(element_name->valuestring), IPFIX_ELEMENT_MAX_LEN - 1));
+        utarray_push_back(elements_array, &element_item);
+    }
+
+    return elements_array;
+error:
+    utarray_free(elements_array);
+    elements_array = NULL;
+    return NULL;
+}
+
+void utable_ipfix_exporter_schema_free(struct ipfix_exporter_schema *instance)
+{
+    if (instance == NULL)
+    {
+        return;
+    }
+
+    if (instance->templates_array)
+    {
+        utarray_free(instance->templates_array);
+        instance->templates_array = NULL;
+    }
+
+    if (instance->worker_context_array)
+    {
+        for (size_t i = 0; i < instance->n_worker; i++)
+        {
+            if (instance->worker_context_array[i].template_blob)
+            {
+                free(instance->worker_context_array[i].template_blob);
+                instance->worker_context_array[i].template_blob = NULL;
+            }
+        }
+    }
+    free(instance->worker_context_array);
+    instance->worker_context_array = NULL;
+
+    free(instance);
+    instance = NULL;
+    return;
+}
+
+struct ipfix_elements_hash
+{
+    char *name;
+    UT_array *elements_array;
+    UT_hash_handle hh;
+};
+
+void ipfix_elements_array_hash_destroy(struct ipfix_elements_hash *elements_array_hash)
+{
+    if (elements_array_hash == NULL)
+    {
+        return;
+    }
+
+    struct ipfix_elements_hash *elements_array_item = NULL;
+    struct ipfix_elements_hash *tmp = NULL;
+    HASH_ITER(hh, elements_array_hash, elements_array_item, tmp)
+    {
+        HASH_DEL(elements_array_hash, elements_array_item);
+        if (elements_array_item->name != NULL)
+        {
+            free(elements_array_item->name);
+            elements_array_item->name = NULL;
+        }
+        if (elements_array_item->elements_array != NULL)
+        {
+            utarray_free(elements_array_item->elements_array);
+            elements_array_item->elements_array = NULL;
+        }
+        free(elements_array_item);
+        elements_array_item = NULL;
+    }
+    return;
+}
+
+struct ipfix_exporter_schema *utable_ipfix_exporter_schema_new(const char *ipfix_schema_json_path, uint16_t domain_id, uint16_t n_worker)
+{
+    if (ipfix_schema_json_path == NULL)
+    {
+        return NULL;
+    }
+
+    size_t json_size = 0;
+    unsigned char *json_str = NULL;
+    int ret = load_file_to_memory(ipfix_schema_json_path, &json_str, &json_size);
+    if (ret == -1)
+    {
+        return NULL;
+    }
+
+    cJSON *root = NULL;
+    root = cJSON_Parse((const char *)json_str);
+    if (root == NULL)
+    {
+        free(json_str);
+        json_str = NULL;
+        return NULL;
+    }
+
+    int version = 0;
+    version = cJSON_GetObjectItem(root, "version")->valueint;
+    if (version < IPFIX_DEFUALT_VERSION)
+    {
+        version = IPFIX_DEFUALT_VERSION;
+    }
+
+    int PEN_number = 0;
+    PEN_number = cJSON_GetObjectItem(root, "PEN_number")->valueint;
+    if (PEN_number <= 0)
+    {
+        PEN_number = IPFIX_DEFUALT_PEN_NUMBER;
+    }
+
+    uint32_t refresh_interval_s = 0;
+    refresh_interval_s = cJSON_GetObjectItem(root, "refresh_interval_s")->valueint;
+    if ((int)refresh_interval_s <= 0)
+    {
+        refresh_interval_s = 60;
+    }
+
+    size_t n_template = 0;
+    cJSON *templates = NULL;
+    templates = cJSON_GetObjectItem(root, "templates");
+    if (templates == NULL || cJSON_GetArraySize(templates) == 0)
+    {
+        free(json_str);
+        json_str = NULL;
+        return NULL;
+    }
+    n_template = cJSON_GetArraySize(templates);
+
+    UT_array *templates_array = NULL;
+    ipfix_utarray_init(&templates_array, sizeof(struct ipfix_template), ipfix_template_item_destroy);
+    struct ipfix_elements_hash *elements_array_hash = NULL;
+
+    for (size_t i = 0; i < n_template; i++)
+    {
+        struct ipfix_template template_item = {0};
+        cJSON *template_obj = cJSON_GetArrayItem(templates, i);
+        cJSON *template_id = cJSON_GetObjectItem(template_obj, "template_id");
+        if (template_id == NULL || template_id->valueint < 256 || template_id->valueint == 0xffff)
+        {
+            goto error;
+        }
+        template_item.template_id = template_id->valueint;
+
+        cJSON *template_name = cJSON_GetObjectItem(template_obj, "template_name");
+        if (template_name == NULL || template_name->valuestring == NULL)
+        {
+            goto error;
+        }
+        template_item.name = (char *)calloc(strlen(template_name->valuestring) + 1, sizeof(char));
+        memcpy(template_item.name, template_name->valuestring, strlen(template_name->valuestring));
+
+        cJSON *elements_key_array = cJSON_GetObjectItem(template_obj, "elements");
+        if (elements_key_array == NULL)
+        {
+            goto error;
+        }
+        size_t n_elements_key = cJSON_GetArraySize(elements_key_array);
+        ipfix_utarray_init(&template_item.elements_array, sizeof(struct ipfix_element), NULL);
+        for (size_t j = 0; j < n_elements_key; j++)
+        {
+            cJSON *element_key = cJSON_GetArrayItem(elements_key_array, j);
+            struct ipfix_elements_hash *elements_array_item = NULL;
+            HASH_FIND_STR(elements_array_hash, element_key->valuestring, elements_array_item);
+            if (elements_array_item == NULL)
+            {
+                elements_array_item = (struct ipfix_elements_hash *)calloc(1, sizeof(struct ipfix_elements_hash));
+                elements_array_item->name = (char *)calloc(strlen(element_key->valuestring) + 1, sizeof(char));
+                memcpy(elements_array_item->name, element_key->valuestring, strlen(element_key->valuestring));
+                HASH_ADD_STR(elements_array_hash, name, elements_array_item);
+                elements_array_item->elements_array = ipfix_elements_array_init(root, element_key->valuestring, PEN_number);
+            }
+
+            if (elements_array_item->elements_array == NULL)
+            {
+                goto error;
+            }
+            utarray_concat(template_item.elements_array, elements_array_item->elements_array); // merge elements_array
+        }
+        utarray_push_back(templates_array, &template_item);
+    }
+    ipfix_elements_array_hash_destroy(elements_array_hash);
+
+    size_t sz_blob = 0;
+    char *template_blob = NULL;
+    ret = ipfix_template_payload_init(templates_array, &template_blob, &sz_blob);
+    if (ret == -1 || sz_blob == 0 || template_blob == NULL)
+    {
+        goto error;
+    }
+
+    struct ipfix_worker_context *worker_context_array = (struct ipfix_worker_context *)calloc(n_worker, sizeof(struct ipfix_worker_context));
+    for (size_t i = 0; i < n_worker; i++)
+    {
+        worker_context_array[i].source_id = (domain_id << 16) | i;
+        worker_context_array[i].sequence = 0;
+        worker_context_array[i].template_blob_length = sz_blob;
+        worker_context_array[i].template_blob = (char *)calloc(sz_blob, sizeof(char));
+        memcpy(worker_context_array[i].template_blob, template_blob, sz_blob);
+
+        struct ipfix_message_head *p_message_head = (struct ipfix_message_head *)(worker_context_array[i].template_blob);
+        p_message_head->version = htons(version);
+        p_message_head->length = htons(sz_blob);
+        p_message_head->domain_id = htonl(worker_context_array[i].source_id);
+    }
+
+    struct ipfix_exporter_schema *_instance = (struct ipfix_exporter_schema *)calloc(1, sizeof(struct ipfix_exporter_schema));
+    _instance->version = version;
+    _instance->PEN_number = PEN_number;
+    _instance->templates_refresh_interval_s = refresh_interval_s;
+    _instance->domain_id = domain_id;
+    _instance->n_worker = n_worker;
+    _instance->templates_array = templates_array;
+    _instance->worker_context_array = worker_context_array;
+
+    free(template_blob);
+    cJSON_Delete(root);
+    free(json_str);
+    json_str = NULL;
+    return _instance;
+
+error:
+    utarray_free(templates_array);
+    cJSON_Delete(root);
+    free(json_str);
+    json_str = NULL;
+    return NULL;
+}
+
+int utable_ipfix_template_get(struct ipfix_exporter_schema *instance, const char *template_name)
+{
+    if (instance == NULL || template_name == NULL)
+    {
+        return -1;
+    }
+
+    size_t n_template = utarray_len(instance->templates_array);
+    for (size_t i = 0; i < n_template; i++)
+    {
+        struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(instance->templates_array, i);
+        if (strcmp(template_item->name, template_name) == 0)
+        {
+            return i;
+        }
+    }
+
+    return -1;
+}
+
+char *ipfix_data_interger_serialize(char *buff, size_t *offset, size_t *buff_sz, int64_t value, size_t sz_value)
+{
+    if (*offset + sz_value >= *buff_sz)
+    {
+        char *new_buff = (char *)realloc(buff, *offset + sz_value + IPFIX_BUFF_MAX_SIZE);
+        if (new_buff == NULL)
+        {
+            free(buff);
+            return NULL;
+        }
+        buff = new_buff;
+        *buff_sz = *offset + sz_value + IPFIX_BUFF_MAX_SIZE;
+    }
+
+    switch (sz_value)
+    {
+        case 1:
+            *(uint8_t *)(buff + *offset) = (uint8_t)value;
+            *offset += 1;
+            break;
+        case 2:
+            *(uint16_t *)(buff + *offset) = htons((uint16_t)value);
+            *offset += 2;
+            break;
+        case 4:
+            *(uint32_t *)(buff + *offset) = htonl((uint32_t)value);
+            *offset += 4;
+            break;
+        case 8:
+            *(uint64_t *)(buff + *offset) = htobe64((uint64_t)value);
+            *offset += 8;
+            break;
+    }
+
+    return buff;
+}
+
+char *ipfix_data_variable_serialize(char *buff, size_t *offset, size_t *buff_sz, char *value, size_t sz_value)
+{
+    if (*offset + sz_value + 3 >= *buff_sz) // 3 means variable_string length > 255, need 3 bytes to describe length
+    {
+        char *new_buff = (char *)realloc(buff, *offset + sz_value + IPFIX_BUFF_MAX_SIZE);
+        if (new_buff == NULL)
+        {
+            free(buff);
+            return NULL;
+        }
+        buff = new_buff;
+        *buff_sz = *offset + sz_value + IPFIX_BUFF_MAX_SIZE;
+    }
+
+    if (sz_value == 0 || value == NULL) // value == NULL need 1 byte to describe length
+    {
+        *(uint8_t *)(buff + *offset) = 0;
+        *offset += 1;
+    }
+    else if (sz_value < 255)
+    {
+        *(uint8_t *)(buff + *offset) = (uint8_t)sz_value;
+        *offset += 1;
+        memcpy(buff + *offset, value, sz_value);
+        *offset += sz_value;
+    }
+    else // >= 255
+    {
+        *(uint8_t *)(buff + *offset) = 255;
+        *offset += 1;
+        *(uint16_t *)(buff + *offset) = htons(sz_value);
+        *offset += 2;
+        memcpy(buff + *offset, value, sz_value);
+        *offset += sz_value;
+    }
+
+    return buff;
+}
+
+int utable_ipfix_data_flow_exporter(const struct utable *table, struct ipfix_exporter_schema *instance, int template_id, uint16_t worker_id, char **blob, size_t *blob_len)
+{
+    if (table == NULL || instance == NULL || worker_id >= instance->n_worker || template_id < 0 || template_id >= (int)utarray_len(instance->templates_array))
+    {
+        return -1;
+    }
+
+    struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(instance->templates_array, (unsigned int)template_id);
+    if (template_item == NULL)
+    {
+        return -1;
+    }
+
+    size_t offset = 0;
+    size_t n_elements = utarray_len(template_item->elements_array);
+    size_t buff_sz = IPFIX_BUFF_MAX_SIZE;
+    char *buff = (char *)calloc(IPFIX_BUFF_MAX_SIZE, sizeof(char));
+    offset += IPFIX_MESSAGE_HEAD_LEN; // skip ipfix_message_head
+
+    *(uint16_t *)(buff + offset) = htons(template_item->template_id); // data_flow_set_id
+    offset += sizeof(uint16_t);                                       // skip data_flow_set_id
+    offset += sizeof(uint16_t);                                       // skip data_flow_set_length
+    for (size_t i = 0; i < n_elements; i++)
+    {
+        struct ipfix_element *p_element = (struct ipfix_element *)utarray_eltptr(template_item->elements_array, i);
+        enum utable_value_type value_type = utable_get_value_type(table, p_element->name);
+        switch (value_type)
+        {
+            case utable_value_type_cstring:
+            case utable_value_type_blob:
+                {
+                    char *value = NULL;
+                    size_t sz_value = 0;
+                    utable_get0_cstring_value(table, p_element->name, &value, &sz_value);
+                    buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, value, sz_value);
+                    break;
+                }
+            case utable_value_type_integer:
+                {
+                    int64_t value = 0;
+                    utable_get0_integer_value(table, p_element->name, &value);
+                    buff = ipfix_data_interger_serialize(buff, &offset, &buff_sz, value, p_element->value_length);
+                    break;
+                }
+            case utable_value_type_integer_array:
+                {
+                    size_t n_array = 0;
+                    int64_t *value_array = NULL;
+                    utable_get0_integer_value_array(table, p_element->name, &value_array, &n_array);
+                    yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL);
+                    yyjson_mut_val *arr = yyjson_mut_arr_with_sint64(doc, value_array, n_array);
+                    yyjson_mut_doc_set_root(doc, arr);
+                    char *value_array_str = yyjson_mut_write(doc, 0, NULL);
+                    if(value_array_str)
+                    {
+                        buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, value_array_str,
+                                                             strlen(value_array_str));
+                        yyjson_mut_doc_free(doc);
+                        free(value_array_str);
+                    }
+                    break;
+                }
+            case utable_value_type_cstring_array:
+                {
+                    size_t n_array = 0;
+                    char **value_array = NULL;
+                    size_t *value_sz = NULL;
+
+                    utable_get0_cstring_value_array(table, p_element->name, &value_array, &value_sz, &n_array);
+                    yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL);
+                    yyjson_mut_val *arr = yyjson_mut_arr_with_strn(doc, (const char **)value_array, value_sz,n_array);
+                    yyjson_mut_doc_set_root(doc, arr);
+                    char *value_array_str = yyjson_mut_write(doc, 0, NULL);
+                    if (value_array_str)
+                    {
+                        buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, value_array_str,
+                                                             strlen(value_array_str));
+                        yyjson_mut_doc_free(doc);
+                        free(value_array_str);
+                    }
+                    break;
+                }                
+            default: // utable_value_type_undefined, ipfix append 0
+                {
+                    switch (p_element->type)
+                    {
+                        case IPFIX_UNSIGNED8:
+                        case IPFIX_UNSIGNED16:
+                        case IPFIX_UNSIGNED32:
+                        case IPFIX_UNSIGNED64:
+                                buff = ipfix_data_interger_serialize(buff, &offset, &buff_sz, 0, p_element->value_length);
+                                break;
+                        case IPFIX_VARIABLE_STRING:
+                                buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, NULL, 0);
+                                break;
+                        default:
+                            {
+                                free(buff);
+                                buff = NULL;
+                                return -1;
+                            }
+                    }
+                    break;
+                }
+        }
+    }
+    *(uint16_t *)(buff + IPFIX_MESSAGE_HEAD_LEN + sizeof(uint16_t)) = htons(offset - IPFIX_MESSAGE_HEAD_LEN); // data_flow_set_length
+
+    struct ipfix_message_head *p_message_head = (struct ipfix_message_head *)(buff);
+    p_message_head->version = htons(instance->version);
+    p_message_head->length = htons(offset);
+    p_message_head->exporttime = htonl(time(NULL)); // time_t is long int, but exporttime is int
+    p_message_head->ipfix_message_sequence = htonl(instance->worker_context_array[worker_id].sequence++);
+    p_message_head->domain_id = htonl(instance->worker_context_array[worker_id].source_id);
+    *blob = buff;
+    *blob_len = offset;
+    return 0;
+}
+
+const char *utable_ipfix_template_flow_get0(struct ipfix_exporter_schema *instance, uint16_t worker_id, size_t *blob_len)
+{
+    if (instance == NULL || instance->worker_context_array == NULL || worker_id >= instance->n_worker)
+    {
+        return NULL;
+    }
+
+    struct ipfix_message_head *p_message_head = (struct ipfix_message_head *)(instance->worker_context_array[worker_id].template_blob);
+
+    p_message_head->exporttime = htonl(time(NULL));                                                     // time_t is long int, but exporttime is int
+    p_message_head->ipfix_message_sequence = htonl(instance->worker_context_array[worker_id].sequence); // template sequence is not increase
+    *blob_len = instance->worker_context_array[worker_id].template_blob_length;
+    return instance->worker_context_array[worker_id].template_blob;
+}
+
+uint32_t utable_ipfix_template_flow_refresh_interval_s(struct ipfix_exporter_schema *ipfix_schema)
+{
+    return ipfix_schema->templates_refresh_interval_s;
+}

deps/utable/version.map

@@ -0,0 +1,6 @@
+{
+  global: 
+	utable_*;
+  local:
+    *;  /* Hide all other symbols */
+};