Adapt to stellar v2.0v2.0.1

18 files changed, 969 insertions, 1165 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 533eea8..1d3c949 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -3,13 +3,14 @@ add_definitions(-fPIC)
 include_directories(/opt/MESA/include/)
 include_directories(/opt/tsg/framework/include/)
 include_directories(${PROJECT_SOURCE_DIR}/deps/)
+include_directories(${CMAKE_BINARY_DIR}/vendor/vbuild/include)
 
 aux_source_directory(${PROJECT_SOURCE_DIR}/deps/mempool DEPS_SRC)
 aux_source_directory(${PROJECT_SOURCE_DIR}/deps/toml DEPS_SRC)
 
-set(HTTP_SRC ${DEPS_SRC} http_decoder.c http_decoder_utils.c http_decoder_half.c
-             http_decoder_table.c http_decoder_string.c http_content_decompress.c
-             http_decoder_result_queue.c)
+set(HTTP_SRC ${DEPS_SRC} http_decoder.cpp http_decoder_utils.cpp http_decoder_half.cpp
+             http_decoder_table.cpp http_decoder_string.cpp http_content_decompress.cpp
+             http_decoder_result_queue.cpp http_decoder_stat.cpp)
 
 add_library(http_decoder SHARED ${HTTP_SRC})
 set_target_properties(http_decoder PROPERTIES LINK_FLAGS "-Wl,--version-script=${PROJECT_SOURCE_DIR}/src/version.map")
diff --git a/src/http_content_decompress.c b/src/http_content_decompress.cpp
index 5fe08a8..e761f1d 100644
--- a/src/http_content_decompress.c
+++ b/src/http_content_decompress.cpp
@@ -7,25 +7,18 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
-
 #include <zlib.h>
 #include <string.h>
 #include <assert.h>
 #include <brotli/decode.h>
-
-#include "stellar/utils.h"
-#include "http_decoder_utils.h"
-#include "http_content_decompress.h"
+#include "http_decoder_inc.h"
 
 #define BUFFER_SIZE (16 * 1024)
 
 struct http_content_decompress {
     enum http_content_encoding encoding;
-
     z_stream *z_stream_ptr;
     BrotliDecoderState *br_state;
-
     char *buffer;
     size_t buffer_size;
 };
@@ -36,15 +29,12 @@ http_content_encoding_str2int(const char *content_encoding)
     if (strcasestr(content_encoding, "gzip") != NULL) {
         return HTTP_CONTENT_ENCODING_GZIP;
     }
-
     if (strcasestr(content_encoding, "deflate") != NULL) {
         return HTTP_CONTENT_ENCODING_DEFLATE;
     }
-
     if (strcasestr(content_encoding, "br") != NULL) {
         return HTTP_CONTENT_ENCODING_BR;
     }
-
     return HTTP_CONTENT_ENCODING_NONE;
 }
 
@@ -54,15 +44,12 @@ http_content_encoding_int2str(enum http_content_encoding content_encoding)
     if (content_encoding == HTTP_CONTENT_ENCODING_GZIP) {
         return "gzip";
     }
-
     if (content_encoding == HTTP_CONTENT_ENCODING_DEFLATE) {
         return "deflate";
     }
-
     if (content_encoding == HTTP_CONTENT_ENCODING_BR) {
         return "br";
     }
-
     return "unknown";
 }
 
@@ -98,7 +85,6 @@ http_content_decompress_create(enum http_content_encoding encoding)
                 goto error;
             }
         }
-
         if (encoding == HTTP_CONTENT_ENCODING_DEFLATE) {
             if (inflateInit2(decompress->z_stream_ptr, -MAX_WBITS) != Z_OK) {
                 goto error;
@@ -112,7 +98,6 @@ http_content_decompress_create(enum http_content_encoding encoding)
             goto error;
         }
     }
-
     return decompress;
 
 error:
@@ -125,17 +110,14 @@ void http_content_decompress_destroy(struct http_content_decompress *decompress)
     if (NULL == decompress) {
         return;
     }
-
     if (decompress->z_stream_ptr != NULL) {
         inflateEnd(decompress->z_stream_ptr);
         FREE(decompress->z_stream_ptr);
     }
-
     if (decompress->br_state) {
         BrotliDecoderDestroyInstance(decompress->br_state);
         decompress->br_state = NULL;
     }
-
     FREE(decompress->buffer);
     FREE(decompress);
 }
@@ -257,12 +239,10 @@ int http_content_decompress_write(struct http_content_decompress *decompress,
         return http_content_decompress_write_zlib(decompress, indata, indata_len,
                                                   outdata, outdata_len);
     }
-
     if (decompress->encoding == HTTP_CONTENT_ENCODING_BR) {
         return http_content_decompress_write_br(decompress, indata, indata_len,
                                                 outdata, outdata_len);
     }
-
     assert(0);
     return -1;
 }
\ No newline at end of file
diff --git a/src/http_decoder.c b/src/http_decoder.c
deleted file mode 100644
index 9dcffe2..0000000
--- a/src/http_decoder.c
+++ /dev/null
@@ -1,991 +0,0 @@
-/*
-**********************************************************************************************
-*	File: http_decoder.c
-*   Description:
-*	Authors: Liu WenTan <[email protected]>
-*	Date:    2024-01-10
-*   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
-***********************************************************************************************
-*/
-
-#include <assert.h>
-#include <stdio.h>
-#include <unistd.h>
-
-#include "toml/toml.h"
-#include "stellar/utils.h"
-#include "stellar/session.h"
-#include "stellar/session_mq.h"
-#include "stellar/session_exdata.h"
-#include "http_decoder.h"
-#include "http_decoder_half.h"
-#include "http_decoder_table.h"
-#include "http_decoder_result_queue.h"
-#include "llhttp.h"
-#include "http_decoder_inc.h"
-#include "fieldstat/fieldstat_easy.h"
-
-#define HTTP_IDENTIFY_LEN 16
-#define HD_RESULT_QUEUE_LEN 16
-
-#define DEFAULT_STAT_OUTPUT_INTERVAL 1
-#define DEFAULT_STAT_INTERVAL_PKTS 1000
-#define DEFAULT_MEMPOOL_SIZE (32 * 1024)
-
-const char *g_hd_cfg_path = "./etc/http/http_decoder.toml";
-const char *http_decoder_topic = "HTTP_DECODER_MESSAGE";
-const char *fs_file_name = "http_decoder.fs";
-
-struct http_decoder_config
-{
-    int decompress_switch;
-    int stat_interval_pkts; // call fieldstat_incrby every stat_interval_pkts
-    int stat_output_interval;
-    size_t result_queue_len; // per session result queue length
-    size_t mempool_size;     // per session mempool size
-};
-
-/**
- * NOTE: http_message don't have the ownership of data
- */
-struct http_message
-{
-    enum http_message_type type;
-    struct http_decoder_result_queue *ref_queue;
-    size_t queue_index;
-};
-
-struct http_decoder
-{
-    struct http_decoder_half *c2s_half;
-    struct http_decoder_half *s2c_half;
-};
-
-struct http_decoder_exdata
-{
-    struct http_decoder_result_queue *queue;
-    struct http_decoder *decoder;
-    nmx_pool_t *mempool;
-};
-
-struct http_decoder_stat
-{
-    long long incoming_bytes;
-    long long incoming_pkts;
-    long long incoming_trans;
-    long long err_pkts;
-    int counter;
-};
-
-struct http_decoder_context
-{
-    int plugin_id;
-    int topic_id;
-    int ex_data_idx;
-    int fs_incoming_bytes_id;
-    int fs_incoming_pkts_id;
-    int fs_incoming_trans_id;
-    int fs_err_pkts_id;
-    struct stellar *st;
-    struct fieldstat_easy *fse;
-    struct http_decoder_config hd_cfg;
-};
-
-__thread struct http_decoder_stat _th_stat;
-
-struct http_message *
-http_message_new(enum http_message_type type,
-                 struct http_decoder_result_queue *queue,
-                 int queue_index)
-{
-    struct http_message *msg = CALLOC(struct http_message, 1);
-
-    msg->type = type;
-    msg->ref_queue = queue;
-    msg->queue_index = queue_index;
-
-    return msg;
-}
-
-static void http_message_free(void *http_msg, void *cb_arg)
-{
-    if (NULL == http_msg)
-    {
-        return;
-    }
-
-    FREE(http_msg);
-}
-
-static void http_event_handler(enum http_event event,
-                               struct http_decoder_half_data **data,
-                               struct http_event_context *ev_ctx)
-{
-    assert(ev_ctx);
-
-    size_t queue_idx = 0;
-    nmx_pool_t *mempool = ev_ctx->ref_mempool;
-    struct http_decoder_result_queue *queue = ev_ctx->ref_queue;
-    struct http_message *msg = NULL;
-    struct http_decoder_half_data *half_data = NULL;
-    int ret = 0;
-
-    switch (event)
-    {
-    case HTTP_EVENT_REQ_INIT:
-        half_data = http_decoder_result_queue_peek_req(queue);
-        if (half_data != NULL)
-        {
-            http_decoder_result_queue_inc_req_index(queue);
-        }
-
-        half_data = http_decoder_result_queue_peek_req(queue);
-        if (half_data != NULL)
-        {
-            half_data = http_decoder_result_queue_pop_req(queue);
-            http_decoder_half_data_free(mempool, half_data);
-            half_data = NULL;
-        }
-
-        half_data = http_decoder_half_data_new(mempool);
-        ret = http_decoder_result_queue_push_req(queue, half_data);
-        if (ret < 0)
-        {
-            fprintf(stderr, "http_decoder_result_queue_push req failed.");
-            http_decoder_half_data_free(mempool, half_data);
-            half_data = NULL;
-        }
-        *data = half_data;
-        break;
-    case HTTP_EVENT_REQ_LINE:
-        queue_idx = http_decoder_result_queue_req_index(queue);
-        msg = http_message_new(HTTP_MESSAGE_REQ_LINE, queue, queue_idx);
-        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
-        break;
-    case HTTP_EVENT_REQ_HDR_END:
-    {
-        int build_url_final = http_decoder_join_url_finally(ev_ctx, http_decoder_result_queue_peek_req(queue), mempool);
-        ret = http_decoder_half_data_has_parsed_header(*data);
-        if (0 == ret && 0 == build_url_final)
-        {
-            break;
-        }
-        queue_idx = http_decoder_result_queue_req_index(queue);
-        msg = http_message_new(HTTP_MESSAGE_REQ_HEADER, queue, queue_idx);
-        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
-    }
-    break;
-    case HTTP_EVENT_REQ_BODY_BEGIN:
-        break;
-    case HTTP_EVENT_REQ_BODY_DATA:
-        queue_idx = http_decoder_result_queue_req_index(queue);
-        msg = http_message_new(HTTP_MESSAGE_REQ_BODY, queue, queue_idx);
-        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
-        break;
-    case HTTP_EVENT_REQ_BODY_END:
-        break;
-    case HTTP_EVENT_REQ_END:
-        http_decoder_result_queue_inc_req_index(queue);
-        half_data = http_decoder_result_queue_pop_req(queue);
-        if (half_data != NULL)
-        {
-            http_decoder_half_data_free(mempool, half_data);
-            half_data = NULL;
-        }
-        break;
-    case HTTP_EVENT_RES_INIT:
-        half_data = http_decoder_result_queue_peek_res(queue);
-        if (half_data != NULL)
-        {
-            http_decoder_result_queue_inc_res_index(queue);
-        }
-
-        half_data = http_decoder_result_queue_peek_res(queue);
-        if (half_data != NULL)
-        {
-            half_data = http_decoder_result_queue_pop_res(queue);
-            http_decoder_half_data_free(mempool, half_data);
-            half_data = NULL;
-        }
-
-        half_data = http_decoder_half_data_new(mempool);
-        ret = http_decoder_result_queue_push_res(queue, half_data);
-        if (ret < 0)
-        {
-            fprintf(stderr, "http_decoder_result_queue_push res failed.");
-            http_decoder_half_data_free(mempool, half_data);
-            half_data = NULL;
-        }
-        *data = half_data;
-        break;
-    case HTTP_EVENT_RES_LINE:
-        queue_idx = http_decoder_result_queue_res_index(queue);
-        msg = http_message_new(HTTP_MESSAGE_RES_LINE, queue, queue_idx);
-        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
-        break;
-    case HTTP_EVENT_RES_HDR:
-        break;
-    case HTTP_EVENT_RES_HDR_END:
-        ret = http_decoder_half_data_has_parsed_header(*data);
-        if (0 == ret)
-        {
-            break;
-        }
-
-        queue_idx = http_decoder_result_queue_res_index(queue);
-        msg = http_message_new(HTTP_MESSAGE_RES_HEADER, queue, queue_idx);
-        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
-        break;
-    case HTTP_EVENT_RES_BODY_BEGIN:
-        break;
-    case HTTP_EVENT_RES_BODY_DATA:
-        queue_idx = http_decoder_result_queue_res_index(queue);
-        msg = http_message_new(HTTP_MESSAGE_RES_BODY, queue, queue_idx);
-        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
-        break;
-    case HTTP_EVENT_RES_BODY_END:
-        break;
-    case HTTP_EVENT_RES_END:
-        http_decoder_result_queue_inc_res_index(queue);
-        half_data = http_decoder_result_queue_pop_res(queue);
-        if (half_data != NULL)
-        {
-            http_decoder_half_data_free(mempool, half_data);
-            half_data = NULL;
-        }
-        break;
-    default:
-        assert(0);
-        break;
-    }
-}
-
-static struct http_decoder *
-http_decoder_new(nmx_pool_t *mempool, http_event_cb *ev_cb,
-                 int decompress_switch)
-{
-    struct http_decoder *decoder = MEMPOOL_CALLOC(mempool, struct http_decoder, 1);
-    assert(decoder);
-
-    decoder->c2s_half = http_decoder_half_new(mempool, ev_cb, HTTP_REQUEST,
-                                              decompress_switch);
-    decoder->s2c_half = http_decoder_half_new(mempool, ev_cb, HTTP_RESPONSE,
-                                              decompress_switch);
-
-    return decoder;
-}
-
-static void
-http_decoder_free(nmx_pool_t *mempool, struct http_decoder *decoder)
-{
-    if (NULL == decoder)
-    {
-        return;
-    }
-
-    if (decoder->c2s_half != NULL)
-    {
-        http_decoder_half_free(mempool, decoder->c2s_half);
-        decoder->c2s_half = NULL;
-    }
-
-    if (decoder->s2c_half != NULL)
-    {
-        http_decoder_half_free(mempool, decoder->s2c_half);
-        decoder->s2c_half = NULL;
-    }
-
-    MEMPOOL_FREE(mempool, decoder);
-}
-
-static struct http_decoder_exdata *
-http_decoder_exdata_new(size_t mempool_size, size_t queue_size,
-                        int decompress_switch)
-{
-    struct http_decoder_exdata *ex_data = CALLOC(struct http_decoder_exdata, 1);
-
-    ex_data->mempool = nmx_create_pool(mempool_size);
-    ex_data->decoder = http_decoder_new(ex_data->mempool, http_event_handler,
-                                        decompress_switch);
-    ex_data->queue = http_decoder_result_queue_new(ex_data->mempool, queue_size);
-
-    return ex_data;
-}
-
-static void http_decoder_exdata_free(struct http_decoder_exdata *ex_data)
-{
-    if (NULL == ex_data)
-    {
-        return;
-    }
-
-    if (ex_data->decoder != NULL)
-    {
-        http_decoder_free(ex_data->mempool, ex_data->decoder);
-        ex_data->decoder = NULL;
-    }
-
-    if (ex_data->queue != NULL)
-    {
-        http_decoder_result_queue_free(ex_data->mempool, ex_data->queue);
-        ex_data->queue = NULL;
-    }
-
-    nmx_destroy_pool(ex_data->mempool);
-
-    FREE(ex_data);
-}
-
-static int http_protocol_identify(const char *data, size_t data_len)
-{
-    llhttp_t parser;
-    llhttp_settings_t settings;
-    enum llhttp_errno error;
-
-    llhttp_settings_init(&settings);
-    llhttp_init(&parser, HTTP_BOTH, &settings);
-
-    error = llhttp_execute(&parser, data, data_len);
-    if (error != HPE_OK)
-    {
-        return -1;
-    }
-
-    return 0;
-}
-
-static int
-http_decoder_stat_init(struct http_decoder_context *ctx, int thread_num)
-{
-    ctx->fse =
-        fieldstat_easy_new(thread_num, "http_decoder_statistics", NULL, 0);
-    if (NULL == ctx->fse)
-    {
-        fprintf(stderr, "fieldstat_easy_new failed.");
-        return -1;
-    }
-
-    ctx->fs_incoming_bytes_id =
-        fieldstat_easy_register_counter(ctx->fse, "incoming_bytes");
-    if (ctx->fs_incoming_bytes_id < 0)
-    {
-        fprintf(stderr, "fieldstat_easy_register_counter incoming_bytes failed.");
-        return -1;
-    }
-
-    ctx->fs_incoming_trans_id =
-        fieldstat_easy_register_counter(ctx->fse, "incoming_trans");
-    if (ctx->fs_incoming_trans_id < 0)
-    {
-        fprintf(stderr, "fieldstat_easy_register_counter incoming_trans failed.");
-        return -1;
-    }
-
-    ctx->fs_incoming_pkts_id =
-        fieldstat_easy_register_counter(ctx->fse, "incoming_pkts");
-    if (ctx->fs_incoming_pkts_id < 0)
-    {
-        fprintf(stderr, "fieldstat_easy_register_counter incoming_pkts failed.");
-        return -1;
-    }
-
-    ctx->fs_err_pkts_id = fieldstat_easy_register_counter(ctx->fse, "err_pkts");
-    if (ctx->fs_err_pkts_id < 0)
-    {
-        fprintf(stderr, "fieldstat_easy_register_counter err_pkts failed.");
-        return -1;
-    }
-
-    int stat_output_interval = DEFAULT_STAT_OUTPUT_INTERVAL;
-    if (ctx->hd_cfg.stat_output_interval > 0)
-    {
-        stat_output_interval = ctx->hd_cfg.stat_output_interval;
-    }
-
-    int ret = fieldstat_easy_enable_auto_output(ctx->fse, fs_file_name,
-                                                stat_output_interval);
-    if (ret < 0)
-    {
-        fprintf(stderr, "fieldstat_easy_enable_auto_output failed.");
-        return -1;
-    }
-
-    sleep(1);
-
-    return 0;
-}
-
-static void
-http_decoder_stat_output(struct http_decoder_context *ctx, int thread_id)
-{
-    if (NULL == ctx || thread_id < 0)
-    {
-        return;
-    }
-
-    int stat_interval_pkts = DEFAULT_STAT_INTERVAL_PKTS;
-    if (ctx->hd_cfg.stat_interval_pkts > 0)
-    {
-        stat_interval_pkts = ctx->hd_cfg.stat_interval_pkts;
-    }
-
-    if (_th_stat.counter >= stat_interval_pkts)
-    {
-        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
-                                      ctx->fs_incoming_bytes_id, NULL, 0,
-                                      _th_stat.incoming_bytes);
-
-        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
-                                      ctx->fs_incoming_pkts_id, NULL, 0,
-                                      _th_stat.incoming_pkts);
-
-        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
-                                      ctx->fs_incoming_trans_id, NULL, 0,
-                                      _th_stat.incoming_trans);
-
-        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
-                                      ctx->fs_err_pkts_id, NULL, 0,
-                                      _th_stat.err_pkts);
-
-        _th_stat.counter = 0;
-        _th_stat.err_pkts = 0;
-        _th_stat.incoming_bytes = 0;
-        _th_stat.incoming_pkts = 0;
-        _th_stat.incoming_trans = 0;
-    }
-}
-
-int http_decoder_entry(struct session *sess, int events,
-                       const struct packet *pkt, void *cb_arg)
-{
-    struct http_decoder_context *ctx = (struct http_decoder_context *)cb_arg;
-    size_t payload_len = 0;
-    uint64_t inner_flag = 0;
-
-    int ret = session_is_inner_most(sess, &inner_flag);
-    if (0 == ret)
-    {
-        return 0;
-    }
-
-    struct http_decoder_exdata *ex_data =
-        session_get_ex_data(sess, ctx->ex_data_idx);
-
-    if (events & SESS_EV_CLOSING)
-    {
-        if (ex_data != NULL)
-        {
-            http_decoder_exdata_free(ex_data);
-            session_set_ex_data(sess, ctx->ex_data_idx, NULL);
-        }
-
-        return 0;
-    }
-
-    const char *payload = session_get0_current_payload(sess, &payload_len);
-
-    if (events & SESS_EV_OPENING)
-    {
-        assert(ex_data == NULL);
-
-        // If not http, ignore this session
-        if (payload_len > 0)
-        {
-            size_t http_identify_len = payload_len > HTTP_IDENTIFY_LEN
-                                           ? HTTP_IDENTIFY_LEN
-                                           : payload_len;
-
-            ret = http_protocol_identify(payload, http_identify_len);
-            if (ret < 0)
-            {
-                // ignore this session's event
-                struct session_event *s_event =
-                    session_get_intrinsic_event(sess, ctx->plugin_id);
-
-                session_event_assign(s_event, ctx->st, sess, 0,
-                                     http_decoder_entry, ctx);
-                return 0;
-            }
-        }
-
-        ex_data = http_decoder_exdata_new(ctx->hd_cfg.mempool_size,
-                                          ctx->hd_cfg.result_queue_len,
-                                          ctx->hd_cfg.decompress_switch);
-        session_set_ex_data(sess, ctx->ex_data_idx, ex_data);
-    }
-
-    if (0 == payload_len || NULL == ex_data)
-    {
-        return 0;
-    }
-
-    int dir = packet_get_direction(pkt);
-    if (dir < 0)
-    {
-        return -1;
-    }
-
-    int thread_id = session_get_current_thread_id(sess);
-    struct http_decoder_half *cur_half = NULL;
-
-    if (dir == PACKET_DIRECTION_C2S)
-    {
-        cur_half = ex_data->decoder->c2s_half;
-    }
-    else
-    {
-        cur_half = ex_data->decoder->s2c_half;
-    }
-
-    http_decoder_half_reinit(cur_half, ctx->topic_id, ex_data->queue,
-                             ex_data->mempool, sess);
-    ret = http_decoder_half_parse(cur_half, payload, payload_len);
-    if (ret < 0)
-    {
-        _th_stat.err_pkts += 1;
-    }
-
-    _th_stat.incoming_bytes += payload_len;
-    _th_stat.incoming_pkts += 1;
-    _th_stat.incoming_trans += http_decoder_half_trans_count(cur_half);
-    _th_stat.counter++;
-
-    http_decoder_stat_output(ctx, thread_id);
-
-    return 0;
-}
-
-static void _http_decoder_context_free(struct http_decoder_context *ctx)
-{
-    if (NULL == ctx)
-    {
-        return;
-    }
-
-    if (ctx->fse != NULL)
-    {
-        fieldstat_easy_free(ctx->fse);
-        ctx->fse = NULL;
-    }
-
-    if (ctx->topic_id >= 0)
-    {
-        session_mq_destroy_topic(ctx->st, ctx->topic_id);
-        ctx->topic_id = -1;
-    }
-
-    FREE(ctx);
-}
-
-static void http_decoder_ex_data_free(struct session *s, int idx,
-                                      void *ex_data, void *arg)
-{
-    if (NULL == ex_data)
-    {
-        return;
-    }
-
-    struct http_decoder_exdata *exdata = (struct http_decoder_exdata *)ex_data;
-    http_decoder_exdata_free(exdata);
-}
-
-static int load_http_decoder_config(const char *cfg_path,
-                                    struct http_decoder_config *hd_cfg)
-{
-    FILE *fp = fopen(cfg_path, "r");
-    if (NULL == fp)
-    {
-        fprintf(stderr, "[%s:%d]Can't open config file:%s",
-                __FUNCTION__, __LINE__, cfg_path);
-        return -1;
-    }
-
-    int ret = 0;
-    char errbuf[256] = {0};
-
-    toml_table_t *root = toml_parse_file(fp, errbuf, sizeof(errbuf));
-    fclose(fp);
-
-    toml_table_t *basic_sec_tbl = toml_table_in(root, "basic");
-    if (NULL == basic_sec_tbl)
-    {
-        fprintf(stderr, "[%s:%d]config file:%s has no key: [basic]",
-                __FUNCTION__, __LINE__, cfg_path);
-        ret = -1;
-        goto next;
-    }
-
-    toml_datum_t int_val = toml_int_in(basic_sec_tbl, "decompress");
-    if (int_val.ok != 0)
-    {
-        hd_cfg->decompress_switch = int_val.u.b;
-    }
-
-    int_val = toml_int_in(basic_sec_tbl, "mempool_size");
-    if (int_val.ok != 0)
-    {
-        hd_cfg->mempool_size = int_val.u.i;
-    }
-    else
-    {
-        hd_cfg->mempool_size = DEFAULT_MEMPOOL_SIZE;
-    }
-
-    int_val = toml_int_in(basic_sec_tbl, "result_queue_len");
-    if (int_val.ok != 0)
-    {
-        hd_cfg->result_queue_len = int_val.u.i;
-    }
-    else
-    {
-        hd_cfg->result_queue_len = HD_RESULT_QUEUE_LEN;
-    }
-
-    int_val = toml_int_in(basic_sec_tbl, "stat_interval_pkts");
-    if (int_val.ok != 0)
-    {
-        hd_cfg->stat_interval_pkts = int_val.u.i;
-    }
-    else
-    {
-        hd_cfg->stat_interval_pkts = DEFAULT_STAT_INTERVAL_PKTS;
-    }
-
-    int_val = toml_int_in(basic_sec_tbl, "stat_output_interval");
-    if (int_val.ok != 0)
-    {
-        hd_cfg->stat_output_interval = int_val.u.i;
-    }
-    else
-    {
-        hd_cfg->stat_output_interval = DEFAULT_STAT_OUTPUT_INTERVAL;
-    }
-
-next:
-    toml_free(root);
-    return ret;
-}
-
-void *http_decoder_init(struct stellar *st)
-{
-    int plugin_id = -1;
-    int topic_id = -1;
-    int thread_num = 0;
-
-    struct http_decoder_context *ctx = CALLOC(struct http_decoder_context, 1);
-
-    int ret = load_http_decoder_config(g_hd_cfg_path, &ctx->hd_cfg);
-    if (ret < 0)
-    {
-        goto failed;
-    }
-
-    ctx->st = st;
-    ctx->ex_data_idx = stellar_session_get_ex_new_index(st, "HTTP_DECODER",
-                                                        http_decoder_ex_data_free,
-                                                        NULL);
-
-    plugin_id = stellar_plugin_register(st, SESS_EV_TCP | SESS_EV_CLOSING,
-                                        http_decoder_entry, ctx);
-    if (plugin_id < 0)
-    {
-        goto failed;
-    }
-
-    ctx->plugin_id = plugin_id;
-
-    topic_id = session_mq_get_topic_id(st, http_decoder_topic);
-    if (topic_id < 0)
-    {
-        topic_id = session_mq_create_topic(st, http_decoder_topic,
-                                           http_message_free, NULL);
-    }
-
-    ctx->topic_id = topic_id;
-
-    thread_num = stellar_get_worker_thread_num(st);
-
-    if (http_decoder_stat_init(ctx, thread_num) < 0)
-    {
-        goto failed;
-    }
-
-    // printf("http_decoder_init: ex_data_idx:%d, plugin_id:%d, topic_id:%d\n",
-    //        ctx->ex_data_idx, ctx->plugin_id, ctx->topic_id);
-    return ctx;
-
-failed:
-    _http_decoder_context_free(ctx);
-    return NULL;
-}
-
-void http_decoder_exit(void *decoder_ctx)
-{
-    if (NULL == decoder_ctx)
-    {
-        return;
-    }
-
-    struct http_decoder_context *ctx =
-        (struct http_decoder_context *)decoder_ctx;
-
-    _http_decoder_context_free(ctx);
-}
-
-enum http_message_type http_message_type(struct http_message *msg)
-{
-    if (NULL == msg)
-    {
-        return HTTP_MESSAGE_MAX;
-    }
-
-    return msg->type;
-}
-
-int http_message_get_request_line(struct http_message *msg,
-                                  struct http_request_line *line)
-{
-    if (NULL == msg || msg->type != HTTP_MESSAGE_REQ_LINE ||
-        NULL == line)
-    {
-        return -1;
-    }
-
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-
-    struct http_decoder_half_data *req_data =
-        msg->ref_queue->array[msg->queue_index].req_data;
-
-    return http_decoder_half_data_get_request_line(req_data, line);
-}
-
-int http_message_get_response_line(struct http_message *msg,
-                                   struct http_response_line *line)
-{
-    if (NULL == msg || msg->type != HTTP_MESSAGE_RES_LINE ||
-        NULL == line)
-    {
-        return -1;
-    }
-
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-
-    struct http_decoder_half_data *res_data =
-        msg->ref_queue->array[msg->queue_index].res_data;
-
-    return http_decoder_half_data_get_response_line(res_data, line);
-}
-
-static int http_msg_get_request_header(struct http_message *msg, struct hstring *key,
-                                       struct http_header *hdr_result)
-{
-    struct http_decoder_half_data *req_data =
-        msg->ref_queue->array[msg->queue_index].req_data;
-    return http_decoder_half_data_get_header(req_data, key, hdr_result);
-}
-
-static int http_msg_get_response_header(struct http_message *msg, struct hstring *key,
-                                        struct http_header *hdr_result)
-{
-    struct http_decoder_half_data *res_data =
-        msg->ref_queue->array[msg->queue_index].res_data;
-    return http_decoder_half_data_get_header(res_data, key, hdr_result);
-}
-
-int http_message_get_header(struct http_message *msg, struct hstring *key,
-                            struct http_header *hdr_result)
-{
-    if (NULL == msg || NULL == key || NULL == hdr_result)
-    {
-        return -1;
-    }
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-    if (HTTP_MESSAGE_REQ_HEADER == msg->type)
-    {
-        return http_msg_get_request_header(msg, key, hdr_result);
-    }
-    else if (HTTP_MESSAGE_RES_HEADER == msg->type)
-    {
-        return http_msg_get_response_header(msg, key, hdr_result);
-    }
-
-    return -1;
-}
-
-static int http_msg_request_header_next(struct http_message *msg,
-                                        struct http_header *hdr)
-{
-    struct http_decoder_half_data *req_data =
-        msg->ref_queue->array[msg->queue_index].req_data;
-    return http_decoder_half_data_iter_header(req_data, hdr);
-}
-
-static int http_msg_response_header_next(struct http_message *msg,
-                                         struct http_header *hdr)
-{
-    struct http_decoder_half_data *res_data =
-        msg->ref_queue->array[msg->queue_index].res_data;
-    return http_decoder_half_data_iter_header(res_data, hdr);
-}
-
-int http_message_header_next(struct http_message *msg,
-                             struct http_header *header)
-{
-    if (NULL == msg || NULL == header)
-    {
-        return -1;
-    }
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-    if (HTTP_MESSAGE_REQ_HEADER == msg->type)
-    {
-        return http_msg_request_header_next(msg, header);
-    }
-    else if (HTTP_MESSAGE_RES_HEADER == msg->type)
-    {
-        return http_msg_response_header_next(msg, header);
-    }
-
-    return -1;
-}
-
-int http_message_reset_header_iter(struct http_message *msg)
-{
-    if (NULL == msg)
-    {
-        return -1;
-    }
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-    if (HTTP_MESSAGE_REQ_HEADER == msg->type)
-    {
-        struct http_decoder_half_data *req_data =
-            msg->ref_queue->array[msg->queue_index].req_data;
-        return http_decoder_half_data_reset_header_iter(req_data);
-    }
-    else if (HTTP_MESSAGE_RES_HEADER == msg->type)
-    {
-        struct http_decoder_half_data *res_data =
-            msg->ref_queue->array[msg->queue_index].res_data;
-        return http_decoder_half_data_reset_header_iter(res_data);
-    }
-
-    return -1;
-}
-
-static int http_msg_get_request_raw_body(struct http_message *msg,
-                                         struct hstring *body)
-{
-    struct http_decoder_half_data *req_data =
-        msg->ref_queue->array[msg->queue_index].req_data;
-    return http_decoder_half_data_get_raw_body(req_data, body);
-}
-
-static int http_msg_get_response_raw_body(struct http_message *msg,
-                                          struct hstring *body)
-{
-    struct http_decoder_half_data *res_data =
-        msg->ref_queue->array[msg->queue_index].res_data;
-    return http_decoder_half_data_get_raw_body(res_data, body);
-}
-
-int http_message_get_raw_body(struct http_message *msg,
-                              struct hstring *body)
-{
-    if (NULL == msg || NULL == body)
-    {
-        return -1;
-    }
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-
-    if (HTTP_MESSAGE_REQ_BODY == msg->type)
-    {
-        return http_msg_get_request_raw_body(msg, body);
-    }
-    else if (HTTP_MESSAGE_RES_BODY == msg->type)
-    {
-        return http_msg_get_response_raw_body(msg, body);
-    }
-
-    return -1;
-}
-
-int http_msg_get_request_decompress_body(struct http_message *msg,
-                                         struct hstring *body)
-{
-    struct http_decoder_half_data *req_data =
-        msg->ref_queue->array[msg->queue_index].req_data;
-    return http_decoder_half_data_get_decompress_body(req_data, body);
-}
-
-int http_msg_get_response_decompress_body(struct http_message *msg,
-                                          struct hstring *body)
-{
-    struct http_decoder_half_data *res_data =
-        msg->ref_queue->array[msg->queue_index].res_data;
-    return http_decoder_half_data_get_decompress_body(res_data, body);
-}
-
-int http_message_get_decompress_body(struct http_message *msg,
-                                     struct hstring *body)
-{
-    if (NULL == msg || NULL == body)
-    {
-        return -1;
-    }
-
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-
-    if (HTTP_MESSAGE_REQ_BODY == msg->type)
-    {
-        return http_msg_get_request_decompress_body(msg, body);
-    }
-    else if (HTTP_MESSAGE_RES_BODY == msg->type)
-    {
-        return http_msg_get_response_decompress_body(msg, body);
-    }
-
-    return -1;
-}
-
-int http_message_get_url(struct http_message *msg, struct hstring *url)
-{
-    if (NULL == msg || NULL == url)
-    {
-        return -1;
-    }
-
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-
-    struct http_decoder_half_data *req_data =
-        msg->ref_queue->array[msg->queue_index].req_data;
-
-    return http_half_data_get_url(req_data, url);
-}
-
-int http_message_get_transaction_seq(struct http_message *msg)
-{
-    if (NULL == msg)
-    {
-        return -1;
-    }
-    assert(msg->ref_queue);
-    assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
-    struct http_decoder_half_data *hf_data;
-    if (HTTP_MESSAGE_REQ_LINE == msg->type || HTTP_MESSAGE_REQ_HEADER == msg->type || HTTP_MESSAGE_REQ_BODY == msg->type)
-    {
-        hf_data = msg->ref_queue->array[msg->queue_index].req_data;
-    }
-    else
-    {
-        hf_data = msg->ref_queue->array[msg->queue_index].res_data;
-    }
-    return http_half_data_get_transaction_seq(hf_data);
-    ;
-}
\ No newline at end of file
diff --git a/src/http_decoder.cpp b/src/http_decoder.cpp
new file mode 100644
index 0000000..f156105
--- /dev/null
+++ b/src/http_decoder.cpp
@@ -0,0 +1,762 @@
+/*
+**********************************************************************************************
+*	File: http_decoder.c
+*   Description:
+*	Authors: Liu WenTan <[email protected]>
+*	Date:    2024-01-10
+*   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
+***********************************************************************************************
+*/
+#include <assert.h>
+#include <stdio.h>
+#include <unistd.h>
+#include "http_decoder_inc.h"
+
+__thread struct http_decoder_stat _th_stat;
+
+struct http_message *http_message_new(enum http_message_type type,
+                                struct http_decoder_result_queue *queue,
+                                int queue_index)
+{
+    struct http_message *msg = CALLOC(struct http_message, 1);
+    msg->type = type;
+    msg->ref_queue = queue;
+    msg->queue_index = queue_index;
+    return msg;
+}
+
+static void http_message_free(void *http_msg, void *cb_arg)
+{
+    if (http_msg)
+    {
+        FREE(http_msg);
+    }
+}
+
+static void http_event_handler(enum http_event event,
+                               struct http_decoder_half_data **data,
+                               struct http_event_context *ev_ctx)
+{
+    assert(ev_ctx);
+
+    size_t queue_idx = 0;
+    nmx_pool_t *mempool = ev_ctx->ref_mempool;
+    struct http_decoder_result_queue *queue = ev_ctx->ref_queue;
+    struct http_message *msg = NULL;
+    struct http_decoder_half_data *half_data = NULL;
+    int ret = 0;
+
+    switch (event)
+    {
+    case HTTP_EVENT_REQ_INIT:
+        half_data = http_decoder_result_queue_peek_req(queue);
+        if (half_data != NULL)
+        {
+            http_decoder_result_queue_inc_req_index(queue);
+        }
+
+        half_data = http_decoder_result_queue_peek_req(queue);
+        if (half_data != NULL)
+        {
+            half_data = http_decoder_result_queue_pop_req(queue);
+            http_decoder_half_data_free(mempool, half_data);
+            half_data = NULL;
+        }
+
+        half_data = http_decoder_half_data_new(mempool);
+        ret = http_decoder_result_queue_push_req(queue, half_data);
+        if (ret < 0)
+        {
+            fprintf(stderr, "http_decoder_result_queue_push req failed.");
+            http_decoder_half_data_free(mempool, half_data);
+            half_data = NULL;
+        }
+        *data = half_data;
+        break;
+    case HTTP_EVENT_REQ_LINE:
+        queue_idx = http_decoder_result_queue_req_index(queue);
+        msg = http_message_new(HTTP_MESSAGE_REQ_LINE, queue, queue_idx);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_REQ_HDR_END:
+    {
+        int build_url_final = http_decoder_join_url_finally(ev_ctx, http_decoder_result_queue_peek_req(queue), mempool);
+        ret = http_decoder_half_data_has_parsed_header(*data);
+        if (0 == ret && 0 == build_url_final)
+        {
+            break;
+        }
+        queue_idx = http_decoder_result_queue_req_index(queue);
+        msg = http_message_new(HTTP_MESSAGE_REQ_HEADER, queue, queue_idx);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+    }
+    break;
+    case HTTP_EVENT_REQ_BODY_BEGIN:
+        break;
+    case HTTP_EVENT_REQ_BODY_DATA:
+        queue_idx = http_decoder_result_queue_req_index(queue);
+        msg = http_message_new(HTTP_MESSAGE_REQ_BODY, queue, queue_idx);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_REQ_BODY_END:
+        break;
+    case HTTP_EVENT_REQ_END:
+        http_decoder_result_queue_inc_req_index(queue);
+        half_data = http_decoder_result_queue_pop_req(queue);
+        if (half_data != NULL)
+        {
+            http_decoder_half_data_free(mempool, half_data);
+            half_data = NULL;
+        }
+        break;
+    case HTTP_EVENT_RES_INIT:
+        half_data = http_decoder_result_queue_peek_res(queue);
+        if (half_data != NULL)
+        {
+            http_decoder_result_queue_inc_res_index(queue);
+        }
+
+        half_data = http_decoder_result_queue_peek_res(queue);
+        if (half_data != NULL)
+        {
+            half_data = http_decoder_result_queue_pop_res(queue);
+            http_decoder_half_data_free(mempool, half_data);
+            half_data = NULL;
+        }
+
+        half_data = http_decoder_half_data_new(mempool);
+        ret = http_decoder_result_queue_push_res(queue, half_data);
+        if (ret < 0)
+        {
+            fprintf(stderr, "http_decoder_result_queue_push res failed.");
+            http_decoder_half_data_free(mempool, half_data);
+            half_data = NULL;
+        }
+        *data = half_data;
+        break;
+    case HTTP_EVENT_RES_LINE:
+        queue_idx = http_decoder_result_queue_res_index(queue);
+        msg = http_message_new(HTTP_MESSAGE_RES_LINE, queue, queue_idx);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_RES_HDR:
+        break;
+    case HTTP_EVENT_RES_HDR_END:
+        ret = http_decoder_half_data_has_parsed_header(*data);
+        if (0 == ret)
+        {
+            break;
+        }
+
+        queue_idx = http_decoder_result_queue_res_index(queue);
+        msg = http_message_new(HTTP_MESSAGE_RES_HEADER, queue, queue_idx);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_RES_BODY_BEGIN:
+        break;
+    case HTTP_EVENT_RES_BODY_DATA:
+        queue_idx = http_decoder_result_queue_res_index(queue);
+        msg = http_message_new(HTTP_MESSAGE_RES_BODY, queue, queue_idx);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_RES_BODY_END:
+        break;
+    case HTTP_EVENT_RES_END:
+        http_decoder_result_queue_inc_res_index(queue);
+        half_data = http_decoder_result_queue_pop_res(queue);
+        if (half_data != NULL)
+        {
+            http_decoder_half_data_free(mempool, half_data);
+            half_data = NULL;
+        }
+        break;
+    default:
+        assert(0);
+        break;
+    }
+}
+
+static struct http_decoder *
+http_decoder_new(nmx_pool_t *mempool, http_event_cb *ev_cb,
+                 int decompress_switch)
+{
+    struct http_decoder *decoder = MEMPOOL_CALLOC(mempool, struct http_decoder, 1);
+    assert(decoder);
+
+    decoder->c2s_half = http_decoder_half_new(mempool, ev_cb, HTTP_REQUEST,
+                                              decompress_switch);
+    decoder->s2c_half = http_decoder_half_new(mempool, ev_cb, HTTP_RESPONSE,
+                                              decompress_switch);
+    return decoder;
+}
+
+static void
+http_decoder_free(nmx_pool_t *mempool, struct http_decoder *decoder)
+{
+    if (NULL == decoder)
+    {
+        return;
+    }
+    if (decoder->c2s_half != NULL)
+    {
+        http_decoder_half_free(mempool, decoder->c2s_half);
+        decoder->c2s_half = NULL;
+    }
+    if (decoder->s2c_half != NULL)
+    {
+        http_decoder_half_free(mempool, decoder->s2c_half);
+        decoder->s2c_half = NULL;
+    }
+    MEMPOOL_FREE(mempool, decoder);
+}
+
+static struct http_decoder_exdata *
+http_decoder_exdata_new(size_t mempool_size, size_t queue_size,
+                        int decompress_switch)
+{
+    struct http_decoder_exdata *ex_data = CALLOC(struct http_decoder_exdata, 1);
+    ex_data->mempool = nmx_create_pool(mempool_size);
+    ex_data->decoder = http_decoder_new(ex_data->mempool, http_event_handler,
+                                        decompress_switch);
+    ex_data->queue = http_decoder_result_queue_new(ex_data->mempool, queue_size);
+
+    return ex_data;
+}
+
+static void http_decoder_exdata_free(struct http_decoder_exdata *ex_data)
+{
+    if (unlikely(NULL == ex_data))
+    {
+        return;
+    }
+    if (ex_data->decoder != NULL)
+    {
+        http_decoder_free(ex_data->mempool, ex_data->decoder);
+        ex_data->decoder = NULL;
+    }
+    if (ex_data->queue != NULL)
+    {
+        http_decoder_result_queue_free(ex_data->mempool, ex_data->queue);
+        ex_data->queue = NULL;
+    }
+
+    nmx_destroy_pool(ex_data->mempool);
+    FREE(ex_data);
+}
+
+static int http_protocol_identify(const char *data, size_t data_len)
+{
+    llhttp_t parser;
+    llhttp_settings_t settings;
+    enum llhttp_errno error;
+
+    llhttp_settings_init(&settings);
+    llhttp_init(&parser, HTTP_BOTH, &settings);
+
+    error = llhttp_execute(&parser, data, data_len);
+    if (error != HPE_OK)
+    {
+        return -1;
+    }
+    return 0;
+}
+
+static void _http_decoder_context_free(struct http_decoder_context *ctx)
+{
+    if (NULL == ctx)
+    {
+        return;
+    }
+
+    if (ctx->fse != NULL)
+    {
+        fieldstat_easy_free(ctx->fse);
+        ctx->fse = NULL;
+    }
+
+    if (ctx->httpd_msg_topic_id >= 0)
+    {
+        stellar_session_mq_destroy_topic(ctx->st, ctx->httpd_msg_topic_id);
+        ctx->httpd_msg_topic_id = -1;
+    }
+
+    FREE(ctx);
+}
+
+static int load_http_decoder_config(const char *cfg_path,
+                                    struct http_decoder_config *hd_cfg)
+{
+    FILE *fp = fopen(cfg_path, "r");
+    if (NULL == fp)
+    {
+        fprintf(stderr, "[%s:%d]Can't open config file:%s",
+                __FUNCTION__, __LINE__, cfg_path);
+        return -1;
+    }
+
+    int ret = 0;
+    char errbuf[256] = {0};
+
+    toml_table_t *root = toml_parse_file(fp, errbuf, sizeof(errbuf));
+    fclose(fp);
+
+    toml_table_t *basic_sec_tbl = toml_table_in(root, "basic");
+    if (NULL == basic_sec_tbl)
+    {
+        fprintf(stderr, "[%s:%d]config file:%s has no key: [basic]",
+                __FUNCTION__, __LINE__, cfg_path);
+        toml_free(root);
+        return -1;
+    }
+
+    toml_datum_t int_val = toml_int_in(basic_sec_tbl, "decompress");
+    if (int_val.ok != 0)
+    {
+        hd_cfg->decompress_switch = int_val.u.b;
+    }
+
+    int_val = toml_int_in(basic_sec_tbl, "mempool_size");
+    if (int_val.ok != 0)
+    {
+        hd_cfg->mempool_size = int_val.u.i;
+    }
+    else
+    {
+        hd_cfg->mempool_size = DEFAULT_MEMPOOL_SIZE;
+    }
+
+    int_val = toml_int_in(basic_sec_tbl, "result_queue_len");
+    if (int_val.ok != 0)
+    {
+        hd_cfg->result_queue_len = int_val.u.i;
+    }
+    else
+    {
+        hd_cfg->result_queue_len = HD_RESULT_QUEUE_LEN;
+    }
+
+    int_val = toml_int_in(basic_sec_tbl, "stat_interval_pkts");
+    if (int_val.ok != 0)
+    {
+        hd_cfg->stat_interval_pkts = int_val.u.i;
+    }
+    else
+    {
+        hd_cfg->stat_interval_pkts = DEFAULT_STAT_INTERVAL_PKTS;
+    }
+
+    int_val = toml_int_in(basic_sec_tbl, "stat_output_interval");
+    if (int_val.ok != 0)
+    {
+        hd_cfg->stat_output_interval = int_val.u.i;
+    }
+    else
+    {
+        hd_cfg->stat_output_interval = DEFAULT_STAT_OUTPUT_INTERVAL;
+    }
+
+    toml_free(root);
+    return ret;
+}
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+    void _httpd_ex_data_free_cb(struct session *s, int idx,
+                                          void *ex_data, void *arg)
+    {
+        if (NULL == ex_data)
+        {
+            return;
+        }
+        struct http_decoder_exdata *exdata = (struct http_decoder_exdata *)ex_data;
+        http_decoder_exdata_free(exdata);
+    }
+
+    void *_httpd_session_ctx_new_cb(struct session *sess, void *plugin_env)
+    {
+        // If not http, ignore this session
+        size_t payload_len;
+        const struct http_decoder_context *httpd_env = (struct http_decoder_context *)plugin_env;
+        const char *payload = session_get0_current_payload(sess, &payload_len);
+        if (payload != NULL && payload_len > 0)
+        {
+            size_t http_identify_len = payload_len > HTTP_IDENTIFY_LEN
+                                           ? HTTP_IDENTIFY_LEN
+                                           : payload_len;
+
+            int ret = http_protocol_identify(payload, http_identify_len);
+            if (ret < 0)
+            {
+                stellar_session_plugin_dettach_current_session(sess);
+                return NULL;
+            }
+        }
+
+        struct http_decoder_exdata *ex_data = http_decoder_exdata_new(httpd_env->hd_cfg.mempool_size,
+                                                                      httpd_env->hd_cfg.result_queue_len,
+                                                                      httpd_env->hd_cfg.decompress_switch);
+        session_exdata_set(sess, httpd_env->ex_data_idx, ex_data);
+        return (void *)"fake_http_decoder_ctx"; // http decoder not use ctx, use exdata only!
+    }
+
+    void _httpd_session_ctx_free_cb(struct session *sess, void *session_ctx, void *plugin_env)
+    {
+        // done in _httpd_ex_data_free_cb()
+    }
+
+    void http_decoder_tcp_stream_msg_cb(struct session *sess, int topic_id, const void *msg, void *no_use_ctx, void *plugin_env)
+    {
+        struct http_decoder_context *httpd_env = (struct http_decoder_context *)plugin_env;
+        struct http_decoder_exdata *ex_data = (struct http_decoder_exdata *)session_exdata_get(sess, httpd_env->ex_data_idx);
+        size_t payload_len;
+
+        const char *payload = session_get0_current_payload(sess, &payload_len);
+        if (unlikely(0 == payload_len || NULL == ex_data))
+        {
+            return;
+        }
+
+        int thread_id = stellar_get_current_thread_id(httpd_env->st);
+        struct http_decoder_half *cur_half = NULL;
+
+        if (PACKET_DIRECTION_C2S == packet_get_direction(session_get0_current_packet(sess)))
+        {
+            cur_half = ex_data->decoder->c2s_half;
+        }
+        else
+        {
+            cur_half = ex_data->decoder->s2c_half;
+        }
+
+        http_decoder_half_reinit(cur_half, httpd_env->httpd_msg_topic_id, ex_data->queue,
+                                 ex_data->mempool, sess);
+        int ret = http_decoder_half_parse(cur_half, payload, payload_len);
+        if (ret < 0)
+        {
+            _th_stat.err_pkts += 1;
+        }
+        _th_stat.incoming_bytes += payload_len;
+        _th_stat.incoming_pkts += 1;
+        _th_stat.incoming_trans += http_decoder_half_trans_count(cur_half);
+        _th_stat.counter++;
+
+        http_decoder_stat_output(httpd_env, thread_id);
+        return;
+    }
+
+    void *http_decoder_init(struct stellar *st)
+    {
+        int httpd_msg_topic_id = -1, tcp_stream_topic_id = -1;
+        int thread_num = 0;
+
+        struct http_decoder_context *httpd_env = CALLOC(struct http_decoder_context, 1);
+        int ret = load_http_decoder_config(HTTPD_CFG_FILE, &httpd_env->hd_cfg);
+        if (ret < 0)
+        {
+            goto failed;
+        }
+        httpd_env->st = st;
+        httpd_env->ex_data_idx = stellar_session_exdata_new_index(st, "HTTP_DECODER",
+                                                                  _httpd_ex_data_free_cb,
+                                                                  NULL);
+        httpd_env->plugin_id = stellar_session_plugin_register(st, _httpd_session_ctx_new_cb,
+                                                               _httpd_session_ctx_free_cb, (void *)httpd_env);
+        if (httpd_env->plugin_id < 0)
+        {
+            goto failed;
+        }
+
+        httpd_msg_topic_id = stellar_session_mq_get_topic_id(st, HTTP_DECODER_TOPIC);
+        if (httpd_msg_topic_id < 0)
+        {
+            httpd_msg_topic_id = stellar_session_mq_create_topic(st, HTTP_DECODER_TOPIC,
+                                                                 http_message_free, NULL);
+        }
+        httpd_env->httpd_msg_topic_id = httpd_msg_topic_id;
+
+        tcp_stream_topic_id = stellar_session_mq_get_topic_id(st, TOPIC_TCP_STREAM);
+        assert(tcp_stream_topic_id >= 0);
+        stellar_session_mq_subscribe(st, tcp_stream_topic_id, http_decoder_tcp_stream_msg_cb, httpd_env->plugin_id);
+
+        thread_num = stellar_get_worker_thread_num(st);
+        assert(thread_num >= 1);
+        if (http_decoder_stat_init(httpd_env, thread_num) < 0)
+        {
+            goto failed;
+        }
+
+        printf("http_decoder_init succ: ex_data_idx:%d, plugin_id:%d, topic_id:%d\n",
+               httpd_env->ex_data_idx, httpd_env->plugin_id, httpd_env->httpd_msg_topic_id);
+        return httpd_env;
+
+    failed:
+        fprintf(stderr, "http_decoder_init fail!\n");
+        _http_decoder_context_free(httpd_env);
+        return NULL;
+    }
+
+    void http_decoder_exit(void *plugin_env)
+    {
+        if (NULL == plugin_env)
+        {
+            return;
+        }
+        struct http_decoder_context *httpd_env = (struct http_decoder_context *)plugin_env;
+        _http_decoder_context_free(httpd_env);
+    }
+
+    enum http_message_type http_message_type(struct http_message *msg)
+    {
+        if (unlikely(NULL == msg))
+        {
+            return HTTP_MESSAGE_MAX;
+        }
+        return msg->type;
+    }
+
+    int http_message_get_request_line(struct http_message *msg,
+                                      struct http_request_line *line)
+    {
+        if (unlikely(NULL == msg || msg->type != HTTP_MESSAGE_REQ_LINE || NULL == line))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+
+        struct http_decoder_half_data *req_data =
+            msg->ref_queue->array[msg->queue_index].req_data;
+
+        return http_decoder_half_data_get_request_line(req_data, line);
+    }
+
+    int http_message_get_response_line(struct http_message *msg,
+                                       struct http_response_line *line)
+    {
+        if (unlikely(NULL == msg || msg->type != HTTP_MESSAGE_RES_LINE || NULL == line))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+
+        struct http_decoder_half_data *res_data =
+            msg->ref_queue->array[msg->queue_index].res_data;
+
+        return http_decoder_half_data_get_response_line(res_data, line);
+    }
+
+    static int http_msg_get_request_header(struct http_message *msg, struct hstring *key,
+                                           struct http_header *hdr_result)
+    {
+        struct http_decoder_half_data *req_data =
+            msg->ref_queue->array[msg->queue_index].req_data;
+        return http_decoder_half_data_get_header(req_data, key, hdr_result);
+    }
+
+    static int http_msg_get_response_header(struct http_message *msg, struct hstring *key,
+                                            struct http_header *hdr_result)
+    {
+        struct http_decoder_half_data *res_data =
+            msg->ref_queue->array[msg->queue_index].res_data;
+        return http_decoder_half_data_get_header(res_data, key, hdr_result);
+    }
+
+    int http_message_get_header(struct http_message *msg, struct hstring *key,
+                                struct http_header *hdr_result)
+    {
+        if (unlikely(NULL == msg || NULL == key || NULL == hdr_result))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        if (HTTP_MESSAGE_REQ_HEADER == msg->type)
+        {
+            return http_msg_get_request_header(msg, key, hdr_result);
+        }
+        else if (HTTP_MESSAGE_RES_HEADER == msg->type)
+        {
+            return http_msg_get_response_header(msg, key, hdr_result);
+        }
+
+        return -1;
+    }
+
+    static int http_msg_request_header_next(struct http_message *msg,
+                                            struct http_header *hdr)
+    {
+        struct http_decoder_half_data *req_data =
+            msg->ref_queue->array[msg->queue_index].req_data;
+        return http_decoder_half_data_iter_header(req_data, hdr);
+    }
+
+    static int http_msg_response_header_next(struct http_message *msg,
+                                             struct http_header *hdr)
+    {
+        struct http_decoder_half_data *res_data =
+            msg->ref_queue->array[msg->queue_index].res_data;
+        return http_decoder_half_data_iter_header(res_data, hdr);
+    }
+
+    int http_message_header_next(struct http_message *msg,
+                                 struct http_header *header)
+    {
+        if (unlikely(NULL == msg || NULL == header))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        if (HTTP_MESSAGE_REQ_HEADER == msg->type)
+        {
+            return http_msg_request_header_next(msg, header);
+        }
+        else if (HTTP_MESSAGE_RES_HEADER == msg->type)
+        {
+            return http_msg_response_header_next(msg, header);
+        }
+
+        return -1;
+    }
+
+    int http_message_reset_header_iter(struct http_message *msg)
+    {
+        if (unlikely(NULL == msg))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        if (HTTP_MESSAGE_REQ_HEADER == msg->type)
+        {
+            struct http_decoder_half_data *req_data =
+                msg->ref_queue->array[msg->queue_index].req_data;
+            return http_decoder_half_data_reset_header_iter(req_data);
+        }
+        else if (HTTP_MESSAGE_RES_HEADER == msg->type)
+        {
+            struct http_decoder_half_data *res_data =
+                msg->ref_queue->array[msg->queue_index].res_data;
+            return http_decoder_half_data_reset_header_iter(res_data);
+        }
+
+        return -1;
+    }
+
+    static int http_msg_get_request_raw_body(struct http_message *msg,
+                                             struct hstring *body)
+    {
+        struct http_decoder_half_data *req_data =
+            msg->ref_queue->array[msg->queue_index].req_data;
+        return http_decoder_half_data_get_raw_body(req_data, body);
+    }
+
+    static int http_msg_get_response_raw_body(struct http_message *msg,
+                                              struct hstring *body)
+    {
+        struct http_decoder_half_data *res_data =
+            msg->ref_queue->array[msg->queue_index].res_data;
+        return http_decoder_half_data_get_raw_body(res_data, body);
+    }
+
+    int http_message_get_raw_body(struct http_message *msg,
+                                  struct hstring *body)
+    {
+        if (unlikely(NULL == msg || NULL == body))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+
+        if (HTTP_MESSAGE_REQ_BODY == msg->type)
+        {
+            return http_msg_get_request_raw_body(msg, body);
+        }
+        else if (HTTP_MESSAGE_RES_BODY == msg->type)
+        {
+            return http_msg_get_response_raw_body(msg, body);
+        }
+
+        return -1;
+    }
+
+    int http_msg_get_request_decompress_body(struct http_message *msg,
+                                             struct hstring *body)
+    {
+        struct http_decoder_half_data *req_data =
+            msg->ref_queue->array[msg->queue_index].req_data;
+        return http_decoder_half_data_get_decompress_body(req_data, body);
+    }
+
+    int http_msg_get_response_decompress_body(struct http_message *msg,
+                                              struct hstring *body)
+    {
+        struct http_decoder_half_data *res_data =
+            msg->ref_queue->array[msg->queue_index].res_data;
+        return http_decoder_half_data_get_decompress_body(res_data, body);
+    }
+
+    int http_message_get_decompress_body(struct http_message *msg,
+                                         struct hstring *body)
+    {
+        if (unlikely(NULL == msg || NULL == body))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+
+        if (HTTP_MESSAGE_REQ_BODY == msg->type)
+        {
+            return http_msg_get_request_decompress_body(msg, body);
+        }
+        else if (HTTP_MESSAGE_RES_BODY == msg->type)
+        {
+            return http_msg_get_response_decompress_body(msg, body);
+        }
+
+        return -1;
+    }
+
+    int http_message_get_url(struct http_message *msg, struct hstring *url)
+    {
+        if (unlikely(NULL == msg || NULL == url))
+        {
+            return -1;
+        }
+
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+
+        struct http_decoder_half_data *req_data =
+            msg->ref_queue->array[msg->queue_index].req_data;
+
+        return http_half_data_get_url(req_data, url);
+    }
+
+    int http_message_get_transaction_seq(struct http_message *msg)
+    {
+        if (unlikely(NULL == msg))
+        {
+            return -1;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        struct http_decoder_half_data *hf_data;
+        if (HTTP_MESSAGE_REQ_LINE == msg->type || HTTP_MESSAGE_REQ_HEADER == msg->type || HTTP_MESSAGE_REQ_BODY == msg->type)
+        {
+            hf_data = msg->ref_queue->array[msg->queue_index].req_data;
+        }
+        else
+        {
+            hf_data = msg->ref_queue->array[msg->queue_index].res_data;
+        }
+        return http_half_data_get_transaction_seq(hf_data);
+    }
+#ifdef __cplusplus
+}
+#endif
\ No newline at end of file
diff --git a/src/http_decoder_half.c b/src/http_decoder_half.cpp
index 7af1956..e2190d7 100644
--- a/src/http_decoder_half.c
+++ b/src/http_decoder_half.cpp
@@ -7,21 +7,12 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
 #include <assert.h>
 #include <stdio.h>
 #include <string.h>
 #include <arpa/inet.h>
-
-#include "stellar/utils.h"
-#include "stellar/session_mq.h"
-#include "llhttp.h"
-#include "http_decoder.h"
 #include "http_decoder_inc.h"
-#include "http_decoder_utils.h"
-#include "http_decoder_half.h"
-#include "http_decoder_table.h"
-#include "http_content_decompress.h"
+#include "llhttp.h"
 
 struct http_decoder_half_data
 {
@@ -205,10 +196,10 @@ static int on_reset(llhttp_t *http)
     return 0;
 }
 
-static inline int is_line_eof(struct http_decoder_half *half)
+static inline int is_line_crlf(struct http_decoder_half *half)
 {
-    const char *chr_r = memrchr(half->data, '\r', half->data_len);
-    const char *chr_n = memrchr(half->data, '\n', half->data_len);
+    const char *chr_r = (char *)memrchr(half->data, '\r', half->data_len);
+    const char *chr_n = (char *)memrchr(half->data, '\n', half->data_len);
     if (chr_r && chr_n && (chr_r + 1 == chr_n))
     {
         return 1;
@@ -238,7 +229,7 @@ static int on_method_complete(llhttp_t *http)
         container_of(http, struct http_decoder_half, parser);
     assert(half);
 
-    if (is_line_eof(half) == 0)
+    if (is_line_crlf(half) == 0)
     {
         http_decoder_table_cache(half->ref_data->table, HTTP_ITEM_METHOD);
     }
@@ -291,7 +282,7 @@ static int on_uri_complete(llhttp_t *http)
         container_of(http, struct http_decoder_half, parser);
     assert(half);
 
-    if (is_line_eof(half) == 0)
+    if (is_line_crlf(half) == 0)
     {
         http_decoder_table_cache(half->ref_data->table, HTTP_ITEM_URI);
     }
@@ -329,7 +320,7 @@ static int on_version_complete(llhttp_t *http)
         container_of(http, struct http_decoder_half, parser);
     assert(half);
 
-    if (is_line_eof(half) == 0)
+    if (is_line_crlf(half) == 0)
     {
         http_decoder_table_cache(half->ref_data->table, HTTP_ITEM_VERSION);
     }
@@ -374,7 +365,7 @@ static int on_status_complete(llhttp_t *http)
         container_of(http, struct http_decoder_half, parser);
     assert(half);
 
-    if (is_line_eof(half) == 0)
+    if (is_line_crlf(half) == 0)
     {
         http_decoder_table_cache(half->ref_data->table, HTTP_ITEM_STATUS);
     }
@@ -614,7 +605,7 @@ static int on_body(llhttp_t *http, const char *at, size_t length)
 
 static void
 http_decoder_half_init(struct http_decoder_half *half,
-                       http_event_cb *http_ev_cb, int type)
+                       http_event_cb *http_ev_cb, enum llhttp_type type)
 {
     if (NULL == half)
     {
@@ -653,14 +644,12 @@ http_decoder_half_init(struct http_decoder_half *half,
     half->settings.on_body = on_body;
 
     half->error = HPE_OK;
-
     half->http_ev_cb = http_ev_cb;
-
     half->ref_data = NULL;
 }
 
 struct http_decoder_half *
-http_decoder_half_new(nmx_pool_t *mempool, http_event_cb *ev_cb, int http_type,
+http_decoder_half_new(nmx_pool_t *mempool, http_event_cb *ev_cb, enum llhttp_type http_type,
                       int decompress_switch)
 {
     struct http_decoder_half *half = MEMPOOL_CALLOC(mempool, struct http_decoder_half, 1);
@@ -752,7 +741,7 @@ int http_decoder_half_parse(struct http_decoder_half *half, const char *data,
     half->error = llhttp_execute(&half->parser, data, data_len);
 
     int ret = 0;
-    uint8_t type = 0;
+    enum llhttp_type type = HTTP_BOTH;
     struct http_decoder_half_data *half_data = NULL;
 
     switch (half->error)
@@ -767,7 +756,7 @@ int http_decoder_half_parse(struct http_decoder_half *half, const char *data,
         ret = 0;
         break;
     default:
-        type = half->parser.type;
+        type = (enum llhttp_type)half->parser.type;
         llhttp_init(&half->parser, type, &half->settings);
         ret = -1;
         break;
@@ -1115,7 +1104,7 @@ int http_decoder_join_url_finally(struct http_event_context *ev_ctx,
 void http_decoder_get_host_feed_url(struct http_decoder_half *half)
 {
     struct http_header host_result = {};
-    struct hstring host_key = {"Host", 4};
+    struct hstring host_key = {(char *)"Host", 4};
     const char *host_refer_str = NULL;
     int host_refer_len = 0;
 
diff --git a/src/http_decoder_half.h b/src/http_decoder_half.h
index a2ff5f8..4e4461b 100644
--- a/src/http_decoder_half.h
+++ b/src/http_decoder_half.h
@@ -7,22 +7,15 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
-
 #ifndef _HTTP_DECODER_HALF_H_
 #define _HTTP_DECODER_HALF_H_
 
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
 #include <stddef.h>
-
 #include "stellar/session.h"
 #include "http_decoder.h"
 #include "http_content_decompress.h"
 #include "http_decoder_result_queue.h"
+#include "llhttp.h"
 
 // only one http event is fired at a time
 enum http_event {
@@ -59,7 +52,7 @@ typedef void http_event_cb(enum http_event event, struct http_decoder_half_data
                            struct http_event_context *ev_ctx);
 
 struct http_decoder_half *
-http_decoder_half_new(nmx_pool_t *mempool, http_event_cb *event_cb, int http_type,
+http_decoder_half_new(nmx_pool_t *mempool, http_event_cb *event_cb, enum llhttp_type http_type,
                       int decompress_switch);
 
 void http_decoder_half_free(nmx_pool_t *mempool, struct http_decoder_half *half);
@@ -111,9 +104,4 @@ int http_decoder_join_url_finally(struct http_event_context *ev_ctx,
                                    nmx_pool_t *mempool);
 int http_half_data_get_url(struct http_decoder_half_data *res_data, struct hstring *url);
 int http_half_data_get_transaction_seq(struct http_decoder_half_data *hf_data);
-
-#ifdef __cplusplus
-}
-#endif
-
 #endif
\ No newline at end of file
diff --git a/src/http_decoder_inc.h b/src/http_decoder_inc.h
index 2d6a5c0..489af3e 100644
--- a/src/http_decoder_inc.h
+++ b/src/http_decoder_inc.h
@@ -15,29 +15,104 @@
 extern "C"
 {
 #endif
+#include "stellar/stellar.h"
+#include "stellar/utils.h"
+#include "stellar/session.h"
+#include "stellar/session_mq.h"
+#include "stellar/session_exdata.h"
 
 #include "mempool/nmx_palloc.h"
 #include "stellar/utils.h"
 #include "http_decoder.h"
 #include "http_decoder_result_queue.h"
+#include "http_decoder_half.h"
+#include "http_decoder_table.h"
+#include "http_decoder_result_queue.h"
+#include "http_decoder_utils.h"
+#include "http_decoder_stat.h"
+#include "fieldstat/fieldstat_easy.h"
+#include "toml/toml.h"
 
+#define likely(x)    __builtin_expect((x), 1)
+#define unlikely(x)  __builtin_expect((x), 0)
 
 #define MEMPOOL_CALLOC(pool, type, number) ((type *)nmx_pcalloc(pool, sizeof(type) * number))
 #define MEMPOOL_REALLOC(pool)
 #define MEMPOOL_FREE(pool, p) nmx_pfree(pool, p)
 
+#define ENABLE_MEMPOOL 1
 #ifdef ENABLE_MEMPOOL
-
 #define HD_CALLOC(pool, type, number) MEMPOOL_CALLOC(pool, number, type)
 #define HD_FREE(pool, p) MEMPOOL_FREE(pool, p)
-
 #else
-
 #define HD_CALLOC(pool, type, number) CALLOC(type, number)
 #define HD_FREE(pool, p) FREE(p)
-
 #endif
 
+#define HTTP_IDENTIFY_LEN 16
+#define HD_RESULT_QUEUE_LEN 16
+
+#define DEFAULT_STAT_OUTPUT_INTERVAL 1
+#define DEFAULT_STAT_INTERVAL_PKTS 1000
+#define DEFAULT_MEMPOOL_SIZE (32 * 1024)
+
+#define HTTPD_CFG_FILE   "./etc/http/http_decoder.toml"
+#define FILEDSTAT_OUTPUT_FILE   "./http_decoder.fs"
+
+struct http_decoder_config
+{
+    int decompress_switch;
+    int stat_interval_pkts; // call fieldstat_incrby every stat_interval_pkts
+    int stat_output_interval;
+    size_t result_queue_len; // per session result queue length
+    size_t mempool_size;     // per session mempool size
+};
+
+/**
+ * NOTE: http_message don't have the ownership of data
+ */
+struct http_message
+{
+    enum http_message_type type;
+    struct http_decoder_result_queue *ref_queue;
+    size_t queue_index;
+};
+
+struct http_decoder
+{
+    struct http_decoder_half *c2s_half;
+    struct http_decoder_half *s2c_half;
+};
+
+struct http_decoder_exdata
+{
+    struct http_decoder_result_queue *queue;
+    struct http_decoder *decoder;
+    nmx_pool_t *mempool;
+};
+
+struct http_decoder_stat
+{
+    long long incoming_bytes;
+    long long incoming_pkts;
+    long long incoming_trans;
+    long long err_pkts;
+    int counter;
+};
+
+struct http_decoder_context
+{
+    int plugin_id;
+    int httpd_msg_topic_id;
+    int ex_data_idx;
+    int fs_incoming_bytes_id;
+    int fs_incoming_pkts_id;
+    int fs_incoming_trans_id;
+    int fs_err_pkts_id;
+    struct stellar *st;
+    struct fieldstat_easy *fse;
+    struct http_decoder_config hd_cfg;
+};
 
 struct http_message;
 
diff --git a/src/http_decoder_result_queue.c b/src/http_decoder_result_queue.cpp
index 9a60d15..32d2895 100644
--- a/src/http_decoder_result_queue.c
+++ b/src/http_decoder_result_queue.cpp
@@ -7,13 +7,8 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
 #include <assert.h>
-
-#include "stellar/utils.h"
-#include "http_decoder_half.h"
 #include "http_decoder_inc.h"
-#include "http_decoder_result_queue.h"
 
 struct http_decoder_result_queue *
 http_decoder_result_queue_new(nmx_pool_t *mempool, size_t queue_size)
@@ -25,11 +20,9 @@ http_decoder_result_queue_new(nmx_pool_t *mempool, size_t queue_size)
     queue->req_index = 0;
     queue->res_index = 0;
     queue->queue_size = queue_size;
-
     queue->array = MEMPOOL_CALLOC(mempool, struct http_decoder_result,
                                   queue->queue_size);
     assert(queue->array);
-
     return queue;
 }
 
@@ -52,17 +45,14 @@ void http_decoder_result_queue_free(nmx_pool_t *mempool,
 				queue->array[i].res_data = NULL;
             }
         }
-        
 		MEMPOOL_FREE(mempool, queue->array);
     }
-
     MEMPOOL_FREE(mempool, queue);
 }
 
 void http_decoder_result_queue_inc_req_index(struct http_decoder_result_queue *queue)
 {
     assert(queue);
-
     queue->req_index++;
     queue->req_index = queue->req_index % queue->queue_size;
 }
@@ -70,7 +60,6 @@ void http_decoder_result_queue_inc_req_index(struct http_decoder_result_queue *q
 void http_decoder_result_queue_inc_res_index(struct http_decoder_result_queue *queue)
 {
     assert(queue);
-
     queue->res_index++;
     queue->res_index = queue->res_index % queue->queue_size;
 }
@@ -78,14 +67,12 @@ void http_decoder_result_queue_inc_res_index(struct http_decoder_result_queue *q
 size_t http_decoder_result_queue_req_index(struct http_decoder_result_queue *queue)
 {
     assert(queue);
-
     return queue->req_index;
 }
 
 size_t http_decoder_result_queue_res_index(struct http_decoder_result_queue *queue)
 {
     assert(queue);
-
     return queue->res_index;
 }
 
@@ -95,12 +82,10 @@ int http_decoder_result_queue_push_req(struct http_decoder_result_queue *queue,
     if (NULL == queue || NULL == req_data) {
         return -1;
     }
-
     assert(queue->array[queue->req_index].req_data == NULL);
     if (queue->array[queue->req_index].req_data != NULL) {
         return -1;
     }
-
     queue->array[queue->req_index].req_data = req_data;
     return 0;
 }
@@ -111,7 +96,6 @@ int http_decoder_result_queue_push_res(struct http_decoder_result_queue *queue,
     if (NULL == queue || NULL == res_data) {
         return -1;
     }
-
     assert(queue->array[queue->res_index].res_data == NULL);
     if (queue->array[queue->res_index].res_data != NULL) {
         return -1;
@@ -127,7 +111,6 @@ http_decoder_result_queue_pop_req(struct http_decoder_result_queue *queue)
     if (NULL == queue) {
         return NULL;
     }
-
     struct http_decoder_half_data *req_data =
         queue->array[queue->req_index].req_data;
     queue->array[queue->req_index].req_data = NULL;
@@ -155,7 +138,6 @@ http_decoder_result_queue_peek_req(struct http_decoder_result_queue *queue)
     if (NULL == queue) {
         return NULL;
     }
-
     assert(queue->req_index < queue->queue_size);
     return queue->array[queue->req_index].req_data;
 }
@@ -166,7 +148,6 @@ http_decoder_result_queue_peek_res(struct http_decoder_result_queue *queue)
     if (NULL == queue) {
         return NULL;
     }
-
     assert(queue->res_index < queue->queue_size);
     return queue->array[queue->res_index].res_data;
 }
\ No newline at end of file
diff --git a/src/http_decoder_result_queue.h b/src/http_decoder_result_queue.h
index cd2163a..4d024eb 100644
--- a/src/http_decoder_result_queue.h
+++ b/src/http_decoder_result_queue.h
@@ -7,18 +7,10 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
-
 #ifndef _HTTP_DECODER_RESULT_QUEUE_H_
 #define _HTTP_DECODER_RESULT_QUEUE_H_
 
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
 #include <stddef.h>
-
 #include "mempool/nmx_palloc.h"
 #include "http_decoder_half.h"
 
@@ -66,8 +58,4 @@ http_decoder_result_queue_peek_req(struct http_decoder_result_queue *queue);
 struct http_decoder_half_data *
 http_decoder_result_queue_peek_res(struct http_decoder_result_queue *queue);
 
-#ifdef __cplusplus
-}
-#endif
-
 #endif
\ No newline at end of file
diff --git a/src/http_decoder_stat.cpp b/src/http_decoder_stat.cpp
new file mode 100644
index 0000000..ffa01eb
--- /dev/null
+++ b/src/http_decoder_stat.cpp
@@ -0,0 +1,99 @@
+#include <assert.h>
+#include <stdio.h>
+#include <unistd.h>
+#include "http_decoder_inc.h"
+
+static __thread struct http_decoder_stat _th_stat;
+
+int http_decoder_stat_init(struct http_decoder_context *ctx, int thread_num)
+{
+    ctx->fse = fieldstat_easy_new(thread_num, "http_decoder_statistics", NULL, 0);
+    if (NULL == ctx->fse)
+    {
+        fprintf(stderr, "fieldstat_easy_new failed.");
+        return -1;
+    }
+
+    ctx->fs_incoming_bytes_id =
+        fieldstat_easy_register_counter(ctx->fse, "incoming_bytes");
+    if (ctx->fs_incoming_bytes_id < 0)
+    {
+        fprintf(stderr, "fieldstat_easy_register_counter incoming_bytes failed.");
+        return -1;
+    }
+
+    ctx->fs_incoming_trans_id =
+        fieldstat_easy_register_counter(ctx->fse, "incoming_trans");
+    if (ctx->fs_incoming_trans_id < 0)
+    {
+        fprintf(stderr, "fieldstat_easy_register_counter incoming_trans failed.");
+        return -1;
+    }
+
+    ctx->fs_incoming_pkts_id =
+        fieldstat_easy_register_counter(ctx->fse, "incoming_pkts");
+    if (ctx->fs_incoming_pkts_id < 0)
+    {
+        fprintf(stderr, "fieldstat_easy_register_counter incoming_pkts failed.");
+        return -1;
+    }
+
+    ctx->fs_err_pkts_id = fieldstat_easy_register_counter(ctx->fse, "err_pkts");
+    if (ctx->fs_err_pkts_id < 0)
+    {
+        fprintf(stderr, "fieldstat_easy_register_counter err_pkts failed.");
+        return -1;
+    }
+
+    int stat_output_interval = DEFAULT_STAT_OUTPUT_INTERVAL;
+    if (ctx->hd_cfg.stat_output_interval > 0)
+    {
+        stat_output_interval = ctx->hd_cfg.stat_output_interval;
+    }
+
+    int ret = fieldstat_easy_enable_auto_output(ctx->fse, FILEDSTAT_OUTPUT_FILE,
+                                                stat_output_interval);
+    if (ret < 0)
+    {
+        fprintf(stderr, "fieldstat_easy_enable_auto_output failed.");
+        return -1;
+    }
+
+    return 0;
+}
+
+void http_decoder_stat_output(struct http_decoder_context *ctx, int thread_id)
+{
+    assert(ctx != NULL);
+
+    int stat_interval_pkts = DEFAULT_STAT_INTERVAL_PKTS;
+    if (ctx->hd_cfg.stat_interval_pkts > 0)
+    {
+        stat_interval_pkts = ctx->hd_cfg.stat_interval_pkts;
+    }
+
+    if (_th_stat.counter >= stat_interval_pkts)
+    {
+        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
+                                      ctx->fs_incoming_bytes_id, NULL, 0,
+                                      _th_stat.incoming_bytes);
+
+        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
+                                      ctx->fs_incoming_pkts_id, NULL, 0,
+                                      _th_stat.incoming_pkts);
+
+        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
+                                      ctx->fs_incoming_trans_id, NULL, 0,
+                                      _th_stat.incoming_trans);
+
+        fieldstat_easy_counter_incrby(ctx->fse, thread_id,
+                                      ctx->fs_err_pkts_id, NULL, 0,
+                                      _th_stat.err_pkts);
+
+        _th_stat.counter = 0;
+        _th_stat.err_pkts = 0;
+        _th_stat.incoming_bytes = 0;
+        _th_stat.incoming_pkts = 0;
+        _th_stat.incoming_trans = 0;
+    }
+}
diff --git a/src/http_decoder_stat.h b/src/http_decoder_stat.h
new file mode 100644
index 0000000..dbf245f
--- /dev/null
+++ b/src/http_decoder_stat.h
@@ -0,0 +1,5 @@
+#ifndef _HTTP_DECODER_STAT_H_
+#define _HTTP_DECODER_STAT_H_ 1
+int http_decoder_stat_init(struct http_decoder_context *ctx, int thread_num);
+void http_decoder_stat_output(struct http_decoder_context *ctx, int thread_id);
+#endif
\ No newline at end of file
diff --git a/src/http_decoder_string.c b/src/http_decoder_string.cpp
index 5414c5b..6744a1e 100644
--- a/src/http_decoder_string.c
+++ b/src/http_decoder_string.cpp
@@ -12,10 +12,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <assert.h>
-
-#include "stellar/utils.h"
-#include "http_decoder_utils.h"
-#include "http_decoder_string.h"
+#include "http_decoder_inc.h"
 
 static const char *string_state_to_desc(enum string_state state)
 {
@@ -64,7 +61,6 @@ static void string_refer2cache(struct http_decoder_string *rstr)
     if (0 == rstr->refer.str_len) {
         return;
     }
-
     if (rstr->cache.str_len >= rstr->max_cache_size) {
         return;
     }
@@ -84,7 +80,6 @@ static void string_refer2cache(struct http_decoder_string *rstr)
     }
 
     rstr->cache.str_len = length;
-
     rstr->refer.str = NULL;
     rstr->refer.str_len = 0;
 }
@@ -93,7 +88,6 @@ static void string_commit2cache(struct http_decoder_string *rstr)
 {
     if (rstr->cache.str_len == rstr->commit.str_len &&
         rstr->cache.str == rstr->commit.str) {
-
         rstr->commit.str = NULL;
         rstr->commit.str_len = 0;
         return;
@@ -141,7 +135,6 @@ void http_decoder_string_cache(struct http_decoder_string *rstr)
         abort();
         break;
     }
-
     rstr->state = STRING_STATE_CACHE;
 }
 
@@ -226,10 +219,8 @@ void http_decoder_string_reinit(struct http_decoder_string *rstr)
 
     rstr->refer.str = NULL;
     rstr->refer.str_len = 0;
-
     rstr->commit.str = NULL;
     rstr->commit.str_len = 0;
-
     rstr->state = STRING_STATE_INIT;
 }
 
diff --git a/src/http_decoder_string.h b/src/http_decoder_string.h
index f9d81dd..ef3a1cc 100644
--- a/src/http_decoder_string.h
+++ b/src/http_decoder_string.h
@@ -11,14 +11,8 @@
 #ifndef _HTTP_DECODER_STRING_H_
 #define _HTTP_DECODER_STRING_H_
 
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
 #include "http_decoder.h"
 
-
 enum string_state {
     STRING_STATE_INIT,
     STRING_STATE_REFER,
@@ -87,9 +81,4 @@ enum string_state http_decoder_string_state(struct http_decoder_string *rstr);
 int http_decoder_string_get(struct http_decoder_string *rstr, struct hstring *out);
 
 void http_decoder_string_dump(struct http_decoder_string *rstr, const char *desc);
-
-#ifdef __cplusplus
-}
-#endif
-
 #endif
\ No newline at end of file
diff --git a/src/http_decoder_table.c b/src/http_decoder_table.cpp
index cc3e7cb..d0f42f0 100644
--- a/src/http_decoder_table.c
+++ b/src/http_decoder_table.cpp
@@ -7,16 +7,10 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
-
 #include <assert.h>
 #include <stdlib.h>
 #include <string.h>
-
-#include "http_decoder_table.h"
-#include "http_decoder.h"
-#include "http_decoder_string.h"
-#include "stellar/utils.h"
+#include "http_decoder_inc.h"
 
 #define INIT_HEADER_CNT 16
 #define MAX_URI_CACHE_SIZE 2048
@@ -90,23 +84,18 @@ void http_decoder_table_free(struct http_decoder_table *table)
     if (NULL == table) {
         return;
     }
-
     if (table->uri.cache.str != NULL) {
         FREE(table->uri.cache.str);
     }
-
     if (table->status.cache.str != NULL) {
         FREE(table->status.cache.str);
     }
-
     if (table->method.cache.str != NULL) {
         FREE(table->method.cache.str);
     }
-
     if (table->version.cache.str != NULL) {
         FREE(table->version.cache.str);
     }
-
     if (table->body.cache.str != NULL) {
         FREE(table->body.cache.str);
     }
@@ -125,7 +114,6 @@ void http_decoder_table_free(struct http_decoder_table *table)
         MEMPOOL_FREE(table->ref_mempool, table->headers);
         table->headers = NULL;
     }
-
     MEMPOOL_FREE(table->ref_mempool, table);
 }
 
@@ -135,7 +123,6 @@ http_decoder_table_state(struct http_decoder_table *table, enum http_item type)
     if (NULL == table) {
         return STRING_STATE_INIT;
     }
-
     struct http_decoder_header *header = NULL;
     enum string_state state = STRING_STATE_INIT;
     assert(table);
@@ -409,7 +396,6 @@ int http_decoder_table_get_uri(struct http_decoder_table *table, struct hstring
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->uri, out);
 }
 
@@ -418,7 +404,6 @@ int http_decoder_table_get_method(struct http_decoder_table *table, struct hstri
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->method, out);
 }
 
@@ -427,7 +412,6 @@ int http_decoder_table_get_status(struct http_decoder_table *table, struct hstri
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->status, out);
 }
 
@@ -436,7 +420,6 @@ int http_decoder_table_get_version(struct http_decoder_table *table, struct hstr
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->version, out);
 }
 
@@ -445,7 +428,6 @@ int http_decoder_table_get_body(struct http_decoder_table *table, struct hstring
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->body, out);
 }
 
@@ -485,7 +467,6 @@ int http_decoder_table_iter_header(struct http_decoder_table *table,
     if (NULL == table || NULL == hdr) {
         return -1;
     }
-
     if (table->header_iter >= table->header_cnt) {
         return -1;
     }
@@ -504,7 +485,6 @@ int http_decoder_table_iter_header(struct http_decoder_table *table,
 
     hdr->key.str = NULL;
     hdr->key.str_len = 0;
-
     hdr->val.str = NULL;
     hdr->val.str_len = 0;
 
@@ -537,7 +517,6 @@ int http_decoder_table_header_complete(struct http_decoder_table *table)
     if (NULL == table) {
         return -1;
     }
-
     return table->header_complete;
 }
 
@@ -546,7 +525,6 @@ void http_decoder_table_set_header_complete(struct http_decoder_table *table)
     if (NULL == table) {
         return;
     }
-
     table->header_complete = 1;
 }
 
@@ -555,6 +533,5 @@ void http_decoder_table_reset_header_complete(struct http_decoder_table *table)
     if (NULL == table) {
         return;
     }
-
     table->header_complete = 0;
 }
\ No newline at end of file
diff --git a/src/http_decoder_table.h b/src/http_decoder_table.h
index 906a28d..de80b9d 100644
--- a/src/http_decoder_table.h
+++ b/src/http_decoder_table.h
@@ -7,17 +7,9 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
-
 #ifndef _HTTP_DECODER_TABLE_H_
 #define _HTTP_DECODER_TABLE_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
 #include <stddef.h>
-
 #include "http_decoder.h"
 #include "http_decoder_inc.h"
 #include "http_decoder_string.h"
@@ -90,9 +82,4 @@ int http_decoder_table_header_complete(struct http_decoder_table *table);
 void http_decoder_table_set_header_complete(struct http_decoder_table *table);
 
 void http_decoder_table_reset_header_complete(struct http_decoder_table *table);
-
-#ifdef __cplusplus
-}
-#endif
-
 #endif
\ No newline at end of file
diff --git a/src/http_decoder_utils.c b/src/http_decoder_utils.cpp
index a5dfbe1..77d716d 100644
--- a/src/http_decoder_utils.c
+++ b/src/http_decoder_utils.cpp
@@ -7,19 +7,15 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
 #include <string.h>
-
-#include "stellar/utils.h"
+#include "http_decoder_inc.h"
 
 char *safe_dup(const char *str, size_t len)
 {
     if (str == NULL || len == 0) {
         return NULL;
     }
-
     char *dup = CALLOC(char, len + 1);
     memcpy(dup, str, len);
-
     return dup;
 }
\ No newline at end of file
diff --git a/src/http_decoder_utils.h b/src/http_decoder_utils.h
index 9c031a3..8cda4ab 100644
--- a/src/http_decoder_utils.h
+++ b/src/http_decoder_utils.h
@@ -7,20 +7,12 @@
 *   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
 ***********************************************************************************************
 */
-
-
 #ifndef _HTTP_DECODER_UTILS_H_
 #define _HTTP_DECODER_UTILS_H_
 
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
 #include <stdlib.h>
 #include <stdio.h>
 
-
 char *safe_dup(const char *str, size_t len);
 
 /******************************************************************************
@@ -58,9 +50,4 @@ enum http_decoder_log_level {
         }                                                                        \
     }
 #endif
-
-#ifdef __cplusplus
-}
-#endif
-
 #endif
\ No newline at end of file
diff --git a/src/version.map b/src/version.map
index 4b8c6d0..be433c2 100644
--- a/src/version.map
+++ b/src/version.map
@@ -3,8 +3,8 @@ global:
 	extern "C" {
         http_message_*;
 		http_decoder_init;
-		http_decoder_entry;
 		http_decoder_exit;
+		http_decoder_tcp_stream_msg_cb;
 	};
 local: *;
 };
\ No newline at end of file