Adapt to stellar2.0;update API,add HDR_END,BODY_END,TRANSACTION_NEW,FREE msg.v2.0.2

19 files changed, 1269 insertions, 1413 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..750a739
--- /dev/null
+++ b/src/http_decoder.cpp
@@ -0,0 +1,833 @@
+#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, uint8_t flow_type)
+{
+    struct http_message *msg = CALLOC(struct http_message, 1);
+    msg->type = type;
+    msg->ref_queue = queue;
+    msg->queue_index = queue_index;
+    msg->flow_type = flow_type;
+    return msg;
+}
+
+static void http_message_free(struct session *sess, 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;
+    u_int8_t flow_flag = 0;
+
+    if(http_event_is_req(event)){
+        queue_idx = http_decoder_result_queue_req_index(queue);
+    }else{
+        queue_idx = http_decoder_result_queue_res_index(queue);
+    }
+
+    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;
+        queue_idx = http_decoder_result_queue_req_index(queue); //get the index after inc
+        msg = http_message_new(HTTP_TRANSACTION_NEW, queue, queue_idx, HTTP_REQUEST);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);        
+        break;
+    case HTTP_EVENT_REQ_LINE:
+        msg = http_message_new(HTTP_MESSAGE_REQ_LINE, queue, queue_idx, HTTP_REQUEST);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_REQ_HDR:
+        msg = http_message_new(HTTP_MESSAGE_REQ_HEADER, queue, queue_idx, HTTP_REQUEST);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+    break;
+    case HTTP_EVENT_REQ_HDR_END:
+    {
+        http_decoder_join_url_finally(ev_ctx, http_decoder_result_queue_peek_req(queue), mempool);
+        /*  maybe some parsed headers in buffer, but has not pushed to plugins yet */
+        half_data = http_decoder_result_queue_peek_req(queue);
+        if(http_decoder_half_data_has_parsed_header(half_data)){
+            msg = http_message_new(HTTP_MESSAGE_REQ_HEADER, queue, queue_idx, HTTP_REQUEST);
+            session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        }
+        http_half_data_update_commit_index(half_data);
+        msg = http_message_new(HTTP_MESSAGE_REQ_HEADER_END, queue, queue_idx, HTTP_REQUEST);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);        
+    }
+    break;
+    case HTTP_EVENT_REQ_BODY_BEGIN:
+        //do nothing, in BODY_DATA event
+        break;
+    case HTTP_EVENT_REQ_BODY_DATA:
+        msg = http_message_new(HTTP_MESSAGE_REQ_BODY, queue, queue_idx, HTTP_REQUEST);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_REQ_BODY_END:
+        msg = http_message_new(HTTP_MESSAGE_REQ_BODY_END, queue, queue_idx, HTTP_REQUEST);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_REQ_END:
+    {
+        if(0 == session_is_symmetric(ev_ctx->ref_session, &flow_flag)){
+            if(SESSION_SEEN_C2S_FLOW == flow_flag){
+                msg = http_message_new(HTTP_TRANSACTION_FREE, queue, queue_idx, HTTP_REQUEST);
+                session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+            }
+        }
+        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;
+        }
+        queue_idx = http_decoder_result_queue_res_index(queue); //get the index after inc
+        *data = half_data;
+        if(0 == session_is_symmetric(ev_ctx->ref_session, &flow_flag)){
+            if(SESSION_SEEN_S2C_FLOW == flow_flag){
+                msg = http_message_new(HTTP_TRANSACTION_NEW, queue, queue_idx, HTTP_RESPONSE);
+                session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);    
+            }
+        }
+        break;
+    case HTTP_EVENT_RES_LINE:
+        msg = http_message_new(HTTP_MESSAGE_RES_LINE, queue, queue_idx, HTTP_RESPONSE);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_RES_HDR:
+        msg = http_message_new(HTTP_MESSAGE_RES_HEADER, queue, queue_idx, HTTP_REQUEST);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_RES_HDR_END:
+        /*  maybe some header in table buffer but has not pushed to plugins */
+        half_data = http_decoder_result_queue_peek_res(queue);
+        if(http_decoder_half_data_has_parsed_header(half_data)){
+            msg = http_message_new(HTTP_MESSAGE_RES_HEADER, queue, queue_idx, HTTP_RESPONSE);
+            session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        }
+        http_half_data_update_commit_index(half_data);
+        msg = http_message_new(HTTP_MESSAGE_RES_HEADER_END, queue, queue_idx, HTTP_RESPONSE);
+        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:
+        msg = http_message_new(HTTP_MESSAGE_RES_BODY, queue, queue_idx, HTTP_RESPONSE);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        break;
+    case HTTP_EVENT_RES_BODY_END:
+        msg = http_message_new(HTTP_MESSAGE_RES_BODY_END, queue, queue_idx, HTTP_RESPONSE);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);    
+        break;
+    case HTTP_EVENT_RES_END:
+        msg = http_message_new(HTTP_TRANSACTION_FREE, queue, queue_idx, HTTP_RESPONSE);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
+        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;
+}
+
+static int http_msg_get_request_header(const struct http_message *msg, const struct hstring *key,
+                                        struct http_header *hdr_result)
+{
+    const 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(const struct http_message *msg, const struct hstring *key,
+                                        struct http_header *hdr_result)
+{
+    const 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);
+}
+
+static int http_msg_request_header_next(const struct http_message *msg,
+                                        struct http_header *hdr)
+{
+    const struct http_decoder_half_data *req_data =
+        msg->ref_queue->array[msg->queue_index].req_data;
+    return http_decoder_half_data_iter_header((struct http_decoder_half_data *)req_data, hdr);
+}
+
+static int http_msg_response_header_next(const struct http_message *msg,
+                                            struct http_header *hdr)
+{
+    const struct http_decoder_half_data *res_data =
+        msg->ref_queue->array[msg->queue_index].res_data;
+    return http_decoder_half_data_iter_header((struct http_decoder_half_data *)res_data, hdr);
+}
+
+static int http_msg_get_request_raw_body(const struct http_message *msg,
+                                            struct hstring *body)
+{
+    const 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(const struct http_message *msg,
+                                            struct hstring *body)
+{
+    const 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);
+}
+
+static int http_msg_get_request_decompress_body(const struct http_message *msg,
+                                            struct hstring *body)
+{
+    const 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);
+}
+
+static int http_msg_get_response_decompress_body(const struct http_message *msg,
+                                            struct hstring *body)
+{
+    const 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);
+}
+
+#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_get(const struct http_message *msg)
+    {
+        if (unlikely(NULL == msg))
+        {
+            return HTTP_MESSAGE_MAX;
+        }
+        return msg->type;
+    }
+
+    void http_message_get_request_line(const struct http_message *msg,
+                                      struct http_request_line *line)
+    {
+        if (unlikely(NULL == msg || msg->type != HTTP_MESSAGE_REQ_LINE))
+        {
+            if(line){ 
+                line->method.str = NULL;
+                line->uri.str = NULL;
+                line->version.str = NULL;
+            }
+            return;
+        }
+        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;
+
+        http_decoder_half_data_get_request_line(req_data, line);
+    }
+
+    void http_message_get_response_line(const struct http_message *msg,
+                                       struct http_response_line *line)
+    {
+        if (unlikely(NULL == msg || msg->type != HTTP_MESSAGE_RES_LINE))
+        {
+            if(line){
+                line->version.str = NULL;
+                line->status.str = NULL;
+            }
+            return;
+        }
+        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;
+
+        http_decoder_half_data_get_response_line(res_data, line);
+    }
+
+    void http_message_get_header(const struct http_message *msg, const struct hstring *key,
+                                struct http_header *hdr_result)
+    {
+        int ret = -1;
+        if (unlikely(NULL == msg || NULL == key))
+        {
+            goto fail;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        if (HTTP_MESSAGE_REQ_HEADER == msg->type)
+        {
+            ret = http_msg_get_request_header(msg, key, hdr_result);
+        }
+        else if (HTTP_MESSAGE_RES_HEADER == msg->type)
+        {
+            ret = http_msg_get_response_header(msg, key, hdr_result);
+        }
+        if(ret >= 0){
+            return;
+        }
+    fail:
+        if(hdr_result){ 
+            hdr_result->key.str = NULL;
+            hdr_result->val.str = NULL;
+        }
+        return;
+    }
+
+    int http_message_header_next(const struct http_message *msg,
+                                 struct http_header *header)
+    {
+        int ret =1;
+        if (unlikely(NULL == msg))
+        {
+            goto fail;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        if (HTTP_MESSAGE_REQ_HEADER == msg->type)
+        {
+            ret = http_msg_request_header_next(msg, header);
+        }
+        else if (HTTP_MESSAGE_RES_HEADER == msg->type)
+        {
+            ret = http_msg_response_header_next(msg, header);
+        }
+        if(ret < 0){
+            goto fail;
+        }
+        return 0;
+    fail:
+        if(header){ 
+            header->key.str = NULL;
+            header->val.str = NULL;
+        }
+        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;
+    }
+
+    void http_message_get_raw_body(const struct http_message *msg,
+                                  struct hstring *body)
+    {
+        int ret = -1;
+        if (unlikely(NULL == msg))
+        {
+            goto fail;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        if (HTTP_MESSAGE_REQ_BODY == msg->type)
+        {
+            ret = http_msg_get_request_raw_body(msg, body);
+        }
+        else if (HTTP_MESSAGE_RES_BODY == msg->type)
+        {
+            ret = http_msg_get_response_raw_body(msg, body);
+        }
+        if(ret < 0){
+            goto fail;
+        }
+        return;
+    fail:
+        if(body){ 
+            body->str = NULL;
+            body->str_len = 0;
+        }
+        return;
+    }
+
+    void http_message_get_decompress_body(const struct http_message *msg,
+                                         struct hstring *body)
+    {
+        int ret = -1;
+        if (unlikely(NULL == msg))
+        {
+            goto fail;
+        }
+        assert(msg->ref_queue);
+        assert(msg->queue_index < HD_RESULT_QUEUE_LEN);
+        if (HTTP_MESSAGE_REQ_BODY == msg->type)
+        {
+            ret = http_msg_get_request_decompress_body(msg, body);
+        }
+        else if (HTTP_MESSAGE_RES_BODY == msg->type)
+        {
+            ret = http_msg_get_response_decompress_body(msg, body);
+        }
+        if(ret < 0){
+            goto fail;
+        }
+        return;
+    fail:
+        if(body){ 
+            body->str = NULL;
+            body->str_len = 0;
+        }
+        return;    
+    }
+
+    void http_message_get_url(const struct http_message *msg, struct hstring *url)
+    {
+        if (unlikely(NULL == msg))
+        {
+            if(url){
+                url->str = NULL;
+                url->str_len = 0;
+            }
+            return;
+        }
+        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;
+
+        if(http_half_data_get_url(req_data, url) < 0){
+            goto fail;
+        }
+        return;
+
+    fail:
+        if(url){
+            url->str = NULL;
+            url->str_len = 0;
+        }
+        return;    
+    }
+
+    int http_message_get_transaction_seq(const 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 = NULL;
+        if(HTTP_REQUEST == msg->flow_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..d3c2ee2 100644
--- a/src/http_decoder_half.c
+++ b/src/http_decoder_half.cpp
@@ -1,27 +1,9 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_half.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 <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
 {
@@ -46,9 +28,9 @@ struct http_decoder_half
     llhttp_t parser;
     llhttp_settings_t settings;
     enum llhttp_errno error;
-
     int decompress_switch;
 
+    // uint8_t is_request_flow;
     enum http_event event;
     http_event_cb *http_ev_cb;
     struct http_event_context *http_ev_ctx;
@@ -79,7 +61,7 @@ static void printf_debug_info(const char *desc, const char *at, size_t length)
     }
 }
 #else
-#define printf_debug_info(desc, at, length)
+#define printf_debug_info(desc, at, length) 
 #endif
 
 static void
@@ -137,7 +119,7 @@ static int on_message_begin(llhttp_t *http)
     half->ref_data = NULL;
 
     assert(half->http_ev_cb != NULL);
-    half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
+    half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx); // http_event_handler()
 
     half->trans_counter++;
     half->ref_data->transaction_index = half->transaction_seq++;
@@ -148,8 +130,7 @@ static int on_message_complete(llhttp_t *http)
 {
     printf_debug_info("on_message_complete", NULL, 0);
 
-    struct http_decoder_half *half =
-        container_of(http, struct http_decoder_half, parser);
+    struct http_decoder_half *half = container_of(http, struct http_decoder_half, parser);
     assert(half);
 
     if (half->parser.type == HTTP_REQUEST)
@@ -157,11 +138,7 @@ static int on_message_complete(llhttp_t *http)
         if (half->event == HTTP_EVENT_REQ_BODY_DATA)
         {
             half->event = HTTP_EVENT_REQ_BODY_END;
-            if (half->http_ev_cb != NULL)
-            {
-                half->http_ev_cb(half->event, &half->ref_data,
-                                 half->http_ev_ctx);
-            }
+            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
         }
     }
     else
@@ -169,11 +146,7 @@ static int on_message_complete(llhttp_t *http)
         if (half->event == HTTP_EVENT_RES_BODY_DATA)
         {
             half->event = HTTP_EVENT_RES_BODY_END;
-            if (half->http_ev_cb != NULL)
-            {
-                half->http_ev_cb(half->event, &half->ref_data,
-                                 half->http_ev_ctx);
-            }
+            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
         }
     }
 
@@ -181,18 +154,12 @@ static int on_message_complete(llhttp_t *http)
     if (half->parser.type == HTTP_REQUEST)
     {
         half->event = HTTP_EVENT_REQ_END;
-        if (half->http_ev_cb != NULL)
-        {
-            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
-        }
+        half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
     }
     else
     {
         half->event = HTTP_EVENT_RES_END;
-        if (half->http_ev_cb != NULL)
-        {
-            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
-        }
+        half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
     }
 
     return 0;
@@ -205,10 +172,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 +205,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 +258,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 +296,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);
     }
@@ -342,7 +309,7 @@ static int on_version_complete(llhttp_t *http)
     if (half->parser.type == HTTP_REQUEST)
     {
         half->event = HTTP_EVENT_REQ_LINE;
-        if (half->http_ev_cb)
+        if (half->http_ev_cb) // http_event_handler()
         {
             half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
         }
@@ -374,7 +341,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);
     }
@@ -385,7 +352,7 @@ static int on_status_complete(llhttp_t *http)
     if (half->parser.type == HTTP_RESPONSE)
     {
         half->event = HTTP_EVENT_RES_LINE;
-        if (half->http_ev_cb != NULL)
+        if (half->http_ev_cb != NULL) // http_event_handler()
         {
             half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
         }
@@ -459,8 +426,7 @@ static int on_header_value_complete(llhttp_t *http)
         struct http_header http_hdr = {0};
         struct hstring key = {.str = (char *)"Content-Encoding", .str_len = 16};
 
-        if (http_decoder_table_get_header(half->ref_data->table, &key,
-                                          &http_hdr) == 0)
+        if (http_decoder_table_get_header(half->ref_data->table, &key, &http_hdr) == 0)
         {
             char encoding_str[MAX_ENCODING_STR_LEN + 1] = {0};
             size_t str_len = http_hdr.val.str_len;
@@ -520,20 +486,12 @@ static int on_headers_complete(llhttp_t *http)
     if (half->parser.type == HTTP_REQUEST)
     {
         half->event = HTTP_EVENT_REQ_HDR_END;
-        if (half->http_ev_cb)
-        {
-            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
-        }
     }
-
-    if (half->parser.type == HTTP_RESPONSE)
+    else
     {
         half->event = HTTP_EVENT_RES_HDR_END;
-        if (half->http_ev_cb)
-        {
-            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
-        }
     }
+    half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);  // http_event_handler()
 
     return 0;
 }
@@ -553,11 +511,7 @@ static int on_body(llhttp_t *http, const char *at, size_t length)
         if (half->event == HTTP_EVENT_REQ_HDR_END)
         {
             half->event = HTTP_EVENT_REQ_BODY_BEGIN;
-            if (half->http_ev_cb)
-            {
-                half->http_ev_cb(half->event, &half->ref_data,
-                                 half->http_ev_ctx);
-            }
+            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx); // http_event_handler()
         }
     }
     else
@@ -565,11 +519,7 @@ static int on_body(llhttp_t *http, const char *at, size_t length)
         if (half->event == HTTP_EVENT_RES_HDR_END)
         {
             half->event = HTTP_EVENT_RES_BODY_BEGIN;
-            if (half->http_ev_cb)
-            {
-                half->http_ev_cb(half->event, &half->ref_data,
-                                 half->http_ev_ctx);
-            }
+            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
         }
     }
 
@@ -595,35 +545,24 @@ static int on_body(llhttp_t *http, const char *at, size_t length)
     if (half->parser.type == HTTP_REQUEST)
     {
         half->event = HTTP_EVENT_REQ_BODY_DATA;
-        if (half->http_ev_cb)
-        {
-            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
-        }
+        half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx); // http_event_handler()
     }
     else
     {
         half->event = HTTP_EVENT_RES_BODY_DATA;
-        if (half->http_ev_cb)
-        {
-            half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
-        }
+        half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);
     }
 
     return 0;
 }
 
-static void
-http_decoder_half_init(struct http_decoder_half *half,
-                       http_event_cb *http_ev_cb, int type)
+static void http_decoder_half_init(struct http_decoder_half *half,
+                       http_event_cb *http_ev_cb, enum llhttp_type type)
 {
-    if (NULL == half)
-    {
-        return;
-    }
-
     llhttp_settings_init(&half->settings);
     llhttp_init(&half->parser, type, &half->settings);
 
+    // half->is_request_flow = (type == HTTP_REQUEST) ? 1 : 0;
     half->settings.on_message_begin = on_message_begin;
     half->settings.on_message_complete = on_message_complete;
     half->settings.on_reset = on_reset;
@@ -653,14 +592,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->http_ev_cb = http_ev_cb;  // http_event_handler()
     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);
@@ -694,7 +631,6 @@ void http_decoder_half_reinit(struct http_decoder_half *half, int topic_id,
                               nmx_pool_t *mempool, struct session *sess)
 {
     assert(half != NULL);
-
     if (half->ref_data != NULL)
     {
         http_decoder_table_reinit(half->ref_data->table);
@@ -712,31 +648,13 @@ static void publish_message_for_parsed_header(struct http_decoder_half *half)
     {
         return;
     }
-
-    // publish complete kv-header message
-    struct http_message *msg = NULL;
-    size_t queue_idx = 0;
-    struct http_decoder_result_queue *queue = half->http_ev_ctx->ref_queue;
-
-    if (half->parser.type == HTTP_REQUEST)
-    {
-        queue_idx = http_decoder_result_queue_req_index(queue);
-
-        msg = http_message_new(HTTP_MESSAGE_REQ_HEADER, queue, queue_idx);
-
-        session_mq_publish_message(half->http_ev_ctx->ref_session,
-                                   half->http_ev_ctx->topic_id, msg);
-    }
-    else
-    {
-        // http response
-        queue_idx = http_decoder_result_queue_res_index(queue);
-
-        msg = http_message_new(HTTP_MESSAGE_RES_HEADER, queue, queue_idx);
-
-        session_mq_publish_message(half->http_ev_ctx->ref_session,
-                                   half->http_ev_ctx->topic_id, msg);
+    if (half->parser.type == HTTP_REQUEST){
+        half->event = HTTP_EVENT_REQ_HDR;
+    }else{
+        half->event = HTTP_EVENT_RES_HDR;
     }
+    half->http_ev_cb(half->event, &half->ref_data, half->http_ev_ctx);  // http_event_handler();
+    return;
 }
 
 int http_decoder_half_parse(struct http_decoder_half *half, const char *data,
@@ -752,7 +670,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 +685,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;
@@ -807,8 +725,9 @@ int http_decoder_half_parse(struct http_decoder_half *half, const char *data,
         {
             http_decoder_table_reset_header_complete(half->ref_data->table);
         }
-        else
+        else 
         {
+            //if headers are not completed with EOF \r\n\r\n, push the parsed headers so far
             publish_message_for_parsed_header(half);
         }
 
@@ -926,11 +845,6 @@ void http_decoder_half_data_free(nmx_pool_t *mempool,
 int http_decoder_half_data_get_request_line(struct http_decoder_half_data *data,
                                             struct http_request_line *line)
 {
-    if (NULL == data || NULL == line)
-    {
-        return -1;
-    }
-
     http_decoder_table_get_method(data->table, &line->method);
     http_decoder_table_get_uri(data->table, &line->uri);
     http_decoder_table_get_version(data->table, &line->version);
@@ -944,11 +858,6 @@ int http_decoder_half_data_get_request_line(struct http_decoder_half_data *data,
 int http_decoder_half_data_get_response_line(struct http_decoder_half_data *data,
                                              struct http_response_line *line)
 {
-    if (NULL == data || NULL == line)
-    {
-        return -1;
-    }
-
     http_decoder_table_get_version(data->table, &line->version);
     http_decoder_table_get_status(data->table, &line->status);
 
@@ -959,25 +868,17 @@ int http_decoder_half_data_get_response_line(struct http_decoder_half_data *data
     return 0;
 }
 
-int http_decoder_half_data_get_header(struct http_decoder_half_data *data,
-                                      struct hstring *key,
+int http_decoder_half_data_get_header(const struct http_decoder_half_data *data,
+                                      const struct hstring *key,
                                       struct http_header *hdr_result)
 {
-    if (NULL == data || NULL == key || NULL == hdr_result)
-    {
-        return -1;
-    }
     return http_decoder_table_get_header(data->table, key, hdr_result);
 }
 
 int http_decoder_half_data_iter_header(struct http_decoder_half_data *data,
                                        struct http_header *header)
 {
-    if (NULL == data || NULL == header)
-    {
-        return -1;
-    }
-    return http_decoder_table_iter_header(data->table, header);
+    return http_decoder_table_iter_header((struct http_decoder_table *)data->table, header);
 }
 
 int http_decoder_half_data_reset_header_iter(struct http_decoder_half_data *req_data)
@@ -998,7 +899,7 @@ int http_decoder_half_data_has_parsed_header(struct http_decoder_half_data *data
     return http_decoder_table_has_parsed_header(data->table);
 }
 
-int http_decoder_half_data_get_raw_body(struct http_decoder_half_data *data,
+int http_decoder_half_data_get_raw_body(const struct http_decoder_half_data *data,
                                         struct hstring *body)
 {
     if (NULL == data || NULL == body)
@@ -1008,14 +909,9 @@ int http_decoder_half_data_get_raw_body(struct http_decoder_half_data *data,
     return http_decoder_table_get_body(data->table, body);
 }
 
-int http_decoder_half_data_get_decompress_body(struct http_decoder_half_data *data,
+int http_decoder_half_data_get_decompress_body(const struct http_decoder_half_data *data,
                                                struct hstring *body)
 {
-    if (NULL == data || NULL == body)
-    {
-        return -1;
-    }
-
     if (HTTP_CONTENT_ENCODING_NONE == data->content_encoding)
     {
         return http_decoder_table_get_body(data->table, body);
@@ -1115,7 +1011,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;
 
@@ -1140,14 +1036,17 @@ int http_half_data_get_url(struct http_decoder_half_data *res_data, struct hstri
     {
         return -1;
     }
-
     url->str = res_data->joint_url.str;
     url->str_len = res_data->joint_url.str_len;
-
     return 0;
 }
 
 int http_half_data_get_transaction_seq(struct http_decoder_half_data *hf_data)
 {
     return hf_data->transaction_index;
+}
+
+void http_half_data_update_commit_index(struct http_decoder_half_data * half_data)
+{
+    http_decoder_table_update_commit_index(half_data->table);
 }
\ No newline at end of file
diff --git a/src/http_decoder_half.h b/src/http_decoder_half.h
index a2ff5f8..362708a 100644
--- a/src/http_decoder_half.h
+++ b/src/http_decoder_half.h
@@ -1,28 +1,12 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_half.h
-*   Description: 
-*	Authors: Liu WenTan <[email protected]>
-*	Date:    2024-01-10
-*   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 +43,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);
@@ -86,18 +70,18 @@ int http_decoder_half_data_get_request_line(struct http_decoder_half_data *data,
 int http_decoder_half_data_get_response_line(struct http_decoder_half_data *data,
                                              struct http_response_line *line);
 
-int http_decoder_half_data_get_header(struct http_decoder_half_data *data,
-                                      struct hstring *key, struct http_header *hdr_res);
+int http_decoder_half_data_get_header(const struct http_decoder_half_data *data,
+                                      const struct hstring *key, struct http_header *hdr_res);
 
 int http_decoder_half_data_iter_header(struct http_decoder_half_data *data,
                                        struct http_header *header);
 int http_decoder_half_data_reset_header_iter(struct http_decoder_half_data *req_data);
 int http_decoder_half_data_has_parsed_header(struct http_decoder_half_data *data);
 
-int http_decoder_half_data_get_raw_body(struct http_decoder_half_data *data,
+int http_decoder_half_data_get_raw_body(const struct http_decoder_half_data *data,
                                         struct hstring *body);
 
-int http_decoder_half_data_get_decompress_body(struct http_decoder_half_data *data,
+int http_decoder_half_data_get_decompress_body(const struct http_decoder_half_data *data,
                                                struct hstring *body);
 
 void http_decoder_half_data_dump(struct http_decoder_half *half);
@@ -112,8 +96,5 @@ int http_decoder_join_url_finally(struct http_event_context *ev_ctx,
 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
-
+void http_half_data_update_commit_index(struct http_decoder_half_data * half_data);
 #endif
\ No newline at end of file
diff --git a/src/http_decoder_inc.h b/src/http_decoder_inc.h
index 2d6a5c0..3e2939d 100644
--- a/src/http_decoder_inc.h
+++ b/src/http_decoder_inc.h
@@ -15,36 +15,114 @@
 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"
 
+#ifndef likely
+#define likely(x)    __builtin_expect((x), 1)
+#endif
+#ifndef unlikely
+#define unlikely(x)  __builtin_expect((x), 0)
+#endif
 
 #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
+{
+    uint8_t flow_type;
+    enum http_message_type type;
+    size_t queue_index;
+    struct http_decoder_result_queue *ref_queue;
+};
+
+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;
 
-struct http_message *
-http_message_new(enum http_message_type type,
-                 struct http_decoder_result_queue *queue,
-                 int queue_index);
+struct http_message *http_message_new(enum http_message_type type, struct http_decoder_result_queue *queue,
+                                    int queue_index, unsigned char flow_type);
 
 #ifdef __cplusplus
 }
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..e10b4a0 100644
--- a/src/http_decoder_string.c
+++ b/src/http_decoder_string.cpp
@@ -1,21 +1,8 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_string.h
-*   Description: 
-*	Authors: LuWenPeng <[email protected]>
-*	Date:    2022-10-31
-*   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
-***********************************************************************************************
-*/
-
 #include <stdio.h>
 #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 +51,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 +70,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 +78,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 +125,6 @@ void http_decoder_string_cache(struct http_decoder_string *rstr)
         abort();
         break;
     }
-
     rstr->state = STRING_STATE_CACHE;
 }
 
@@ -226,19 +209,17 @@ 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;
 }
 
-enum string_state http_decoder_string_state(struct http_decoder_string *rstr)
+enum string_state http_decoder_string_state(const struct http_decoder_string *rstr)
 {
     return rstr->state;
 }
 
-int http_decoder_string_get(struct http_decoder_string *rstr, struct hstring *out)
+int http_decoder_string_get(const struct http_decoder_string *rstr, struct hstring *out)
 {
     if (NULL == rstr || NULL == out) {
         return -1;
diff --git a/src/http_decoder_string.h b/src/http_decoder_string.h
index f9d81dd..4c95960 100644
--- a/src/http_decoder_string.h
+++ b/src/http_decoder_string.h
@@ -1,24 +1,8 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_string.h
-*   Description: 
-*	Authors: LuWenPeng <[email protected]>
-*	Date:    2022-10-31
-*   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
-***********************************************************************************************
-*/
-
 #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,
@@ -82,14 +66,9 @@ void http_decoder_string_init(struct http_decoder_string *rstr,
 
 void http_decoder_string_reinit(struct http_decoder_string *rstr);
 
-enum string_state http_decoder_string_state(struct http_decoder_string *rstr);
+enum string_state http_decoder_string_state(const struct http_decoder_string *rstr);
 
-int http_decoder_string_get(struct http_decoder_string *rstr, struct hstring *out);
+int http_decoder_string_get(const 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..b11e1b9 100644
--- a/src/http_decoder_table.c
+++ b/src/http_decoder_table.cpp
@@ -1,22 +1,7 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_table.c
-*   Description:
-*	Authors: LuWenPeng <[email protected]>
-*	Date:    2022-10-31
-*   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
@@ -41,8 +26,9 @@ struct http_decoder_table {
     nmx_pool_t *ref_mempool;
     int header_complete; // flag for all headers parsed completely
     size_t header_cnt;
-    size_t header_index;
-    size_t header_iter;
+    size_t header_index; //current parsing header
+    size_t header_iter; //plugins iterate cursor
+    size_t commit_header_index; //pushed to plugins, whether has called http_message_header_next()
     struct http_decoder_header *headers;
 };
 
@@ -79,7 +65,7 @@ struct http_decoder_table *http_decoder_table_new(nmx_pool_t *mempool)
     table->header_cnt = INIT_HEADER_CNT;
     table->headers = MEMPOOL_CALLOC(mempool, struct http_decoder_header,
                                     table->header_cnt);
-
+    table->commit_header_index = 0;
     http_decoder_table_init(table);
 
     return table;
@@ -90,23 +76,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 +106,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 +115,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);
@@ -361,18 +340,16 @@ void http_decoder_table_reset(struct http_decoder_table *table, enum http_item t
 
 void http_decoder_table_reinit(struct http_decoder_table *table)
 {
-    if (NULL == table) {
-        return;
-    }
-
-    struct http_decoder_header *header = NULL;
     assert(table);
-
+    struct http_decoder_header *header = NULL;
+    
     http_decoder_string_reinit(&table->uri);
     http_decoder_string_reinit(&table->status);
     http_decoder_string_reinit(&table->method);
     http_decoder_string_reinit(&table->version);
-    for (size_t i = 0; i < table->header_iter; i++) {
+    // for (size_t i = 0; i < table->header_iter; i++) {
+    for (size_t i = 0; i < table->commit_header_index; i++) {
+        //todo, reset header_index, avoid realloc headers as much as possible
         header = &table->headers[i];
         http_decoder_string_reinit(&header->key);
         http_decoder_string_reinit(&header->val);
@@ -404,60 +381,51 @@ void http_decoder_table_dump(struct http_decoder_table *table)
     }
 }
 
-int http_decoder_table_get_uri(struct http_decoder_table *table, struct hstring *out)
+int http_decoder_table_get_uri(const struct http_decoder_table *table, struct hstring *out)
 {
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->uri, out);
 }
 
-int http_decoder_table_get_method(struct http_decoder_table *table, struct hstring *out)
+int http_decoder_table_get_method(const struct http_decoder_table *table, struct hstring *out)
 {
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->method, out);
 }
 
-int http_decoder_table_get_status(struct http_decoder_table *table, struct hstring *out)
+int http_decoder_table_get_status(const struct http_decoder_table *table, struct hstring *out)
 {
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->status, out);
 }
 
-int http_decoder_table_get_version(struct http_decoder_table *table, struct hstring *out)
+int http_decoder_table_get_version(const struct http_decoder_table *table, struct hstring *out)
 {
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->version, out);
 }
 
-int http_decoder_table_get_body(struct http_decoder_table *table, struct hstring *out)
+int http_decoder_table_get_body(const struct http_decoder_table *table, struct hstring *out)
 {
     if (NULL == table || NULL == out) {
         return -1;
     }
-
     return http_decoder_string_get(&table->body, out);
 }
 
-int http_decoder_table_get_header(struct http_decoder_table *table, struct hstring *key,
+int http_decoder_table_get_header(const struct http_decoder_table *table, const struct hstring *key,
                                   struct http_header *hdr_result)
 {
-    if (NULL == table || NULL == key->str || 0 == key->str_len || NULL == hdr_result) {
-        return -1;
-    }
-
     for (size_t i = 0; i < table->header_cnt; i++) {
-        struct http_decoder_header *tmp_header = &table->headers[i];
+        const struct http_decoder_header *tmp_header = &table->headers[i];
         if (tmp_header->key.commit.str_len != key->str_len) {
             continue;
         }
@@ -475,7 +443,6 @@ int http_decoder_table_get_header(struct http_decoder_table *table, struct hstri
             }
         }
     }
-
     return -1;
 }
 
@@ -485,7 +452,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;
     }
@@ -498,13 +464,12 @@ int http_decoder_table_iter_header(struct http_decoder_table *table,
             http_decoder_string_get(&tmp_header->key, &hdr->key);
             http_decoder_string_get(&tmp_header->val, &hdr->val);
             table->header_iter++;
-            return 0;
+            return 1;
         }
     }
 
     hdr->key.str = NULL;
     hdr->key.str_len = 0;
-
     hdr->val.str = NULL;
     hdr->val.str_len = 0;
 
@@ -519,13 +484,15 @@ int http_decoder_table_reset_header_iter(struct http_decoder_table *table)
 
 int http_decoder_table_has_parsed_header(struct http_decoder_table *table)
 {
-    if (NULL == table || (table->header_iter == table->header_index)) {
+    // if (NULL == table || (table->header_iter == table->header_index)) {
+    if (NULL == table || (table->commit_header_index == table->header_index)) {
         return 0;
     }
 
-    struct http_decoder_header *tmp_header = &table->headers[table->header_iter];
-    if (http_decoder_string_state(&tmp_header->key) == STRING_STATE_COMMIT &&
-        http_decoder_string_state(&tmp_header->val) == STRING_STATE_COMMIT) {
+    const struct http_decoder_header *tmp_header = &table->headers[table->header_iter];
+
+    if (http_decoder_string_state(&tmp_header->key) == STRING_STATE_COMMIT 
+         && http_decoder_string_state(&tmp_header->val) == STRING_STATE_COMMIT) {
         return 1;
     }
 
@@ -537,7 +504,6 @@ int http_decoder_table_header_complete(struct http_decoder_table *table)
     if (NULL == table) {
         return -1;
     }
-
     return table->header_complete;
 }
 
@@ -546,7 +512,6 @@ void http_decoder_table_set_header_complete(struct http_decoder_table *table)
     if (NULL == table) {
         return;
     }
-
     table->header_complete = 1;
 }
 
@@ -555,6 +520,10 @@ void http_decoder_table_reset_header_complete(struct http_decoder_table *table)
     if (NULL == table) {
         return;
     }
-
     table->header_complete = 0;
+}
+
+void http_decoder_table_update_commit_index(struct http_decoder_table *table)
+{
+    table->commit_header_index = table->header_index;
 }
\ No newline at end of file
diff --git a/src/http_decoder_table.h b/src/http_decoder_table.h
index 906a28d..ce79d46 100644
--- a/src/http_decoder_table.h
+++ b/src/http_decoder_table.h
@@ -1,23 +1,6 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_table.h
-*   Description:
-*	Authors: LuWenPeng <[email protected]>
-*	Date:    2022-10-31
-*   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"
@@ -53,18 +36,18 @@ void http_decoder_table_reinit(struct http_decoder_table *table);
 
 void http_decoder_table_dump(struct http_decoder_table *table);
 
-int http_decoder_table_get_uri(struct http_decoder_table *table, struct hstring *out);
+int http_decoder_table_get_uri(const struct http_decoder_table *table, struct hstring *out);
 
-int http_decoder_table_get_method(struct http_decoder_table *table, struct hstring *out);
+int http_decoder_table_get_method(const struct http_decoder_table *table, struct hstring *out);
 
-int http_decoder_table_get_status(struct http_decoder_table *table, struct hstring *out);
+int http_decoder_table_get_status(const struct http_decoder_table *table, struct hstring *out);
 
-int http_decoder_table_get_version(struct http_decoder_table *table, struct hstring *out);
+int http_decoder_table_get_version(const struct http_decoder_table *table, struct hstring *out);
 
-int http_decoder_table_get_body(struct http_decoder_table *table, struct hstring *out);
+int http_decoder_table_get_body(const struct http_decoder_table *table, struct hstring *out);
 
-int http_decoder_table_get_header(struct http_decoder_table *table,
-                                  struct hstring *key,
+int http_decoder_table_get_header(const struct http_decoder_table *table,
+                                  const struct hstring *key,
                                   struct http_header *hdr_res);
 
 int http_decoder_table_iter_header(struct http_decoder_table *table,
@@ -91,8 +74,6 @@ 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
+void http_decoder_table_update_commit_index(struct http_decoder_table *table);
 
 #endif
\ No newline at end of file
diff --git a/src/http_decoder_utils.c b/src/http_decoder_utils.c
deleted file mode 100644
index a5dfbe1..0000000
--- a/src/http_decoder_utils.c
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_utils.c
-*   Description: 
-*	Authors: LuWenPeng <[email protected]>
-*	Date:    2022-10-31
-*   Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved.
-***********************************************************************************************
-*/
-
-#include <string.h>
-
-#include "stellar/utils.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.cpp b/src/http_decoder_utils.cpp
new file mode 100644
index 0000000..5686e2d
--- /dev/null
+++ b/src/http_decoder_utils.cpp
@@ -0,0 +1,137 @@
+#include <string.h>
+#include <assert.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;
+}
+
+const char *http_message_type_to_string(enum http_message_type type)
+{
+    const char *sname = "unknown_msg_type";
+
+    switch (type)
+    {
+    case HTTP_MESSAGE_REQ_LINE:
+        sname = "HTTP_MESSAGE_REQ_LINE";
+        break;
+    case HTTP_MESSAGE_REQ_HEADER:
+        sname = "HTTP_MESSAGE_REQ_HEADER";
+        break;
+    case HTTP_MESSAGE_REQ_HEADER_END:
+        sname = "HTTP_MESSAGE_REQ_HEADER_END";
+        break;
+    case HTTP_MESSAGE_REQ_BODY: 
+        sname = "HTTP_MESSAGE_REQ_BODY";
+        break;
+    case HTTP_MESSAGE_REQ_BODY_END:
+        sname = "HTTP_MESSAGE_REQ_BODY_END";
+        break;
+    case HTTP_MESSAGE_RES_LINE:
+        sname = "HTTP_MESSAGE_RES_LINE";
+        break;  
+    case HTTP_MESSAGE_RES_HEADER:       
+        sname = "HTTP_MESSAGE_RES_HEADER";
+        break;
+    case HTTP_MESSAGE_RES_HEADER_END:
+        sname = "HTTP_MESSAGE_RES_HEADER_END";
+        break;
+    case HTTP_MESSAGE_RES_BODY:
+        sname = "HTTP_MESSAGE_RES_BODY";
+        break;  
+    case HTTP_MESSAGE_RES_BODY_END:
+        sname = "HTTP_MESSAGE_RES_BODY_END";
+        break;
+    case HTTP_TRANSACTION_NEW:
+        sname = "HTTP_TRANSACTION_NEW";
+        break;
+    case HTTP_TRANSACTION_FREE:
+        sname = "HTTP_TRANSACTION_FREE";
+        break;
+
+    default:
+        break;
+    }
+
+    return sname;
+}
+
+int http_message_type_is_req(struct session *sess, enum http_message_type msg_type)
+{
+    int is_req_msg = 0;
+
+    switch(msg_type){
+        case HTTP_MESSAGE_REQ_LINE:
+        case HTTP_MESSAGE_REQ_HEADER:
+        case HTTP_MESSAGE_REQ_HEADER_END:
+        case HTTP_MESSAGE_REQ_BODY:
+        case HTTP_MESSAGE_REQ_BODY_END:
+            is_req_msg = 1;
+        break;
+
+        case HTTP_MESSAGE_RES_LINE:
+        case HTTP_MESSAGE_RES_HEADER:
+        case HTTP_MESSAGE_RES_HEADER_END:
+        case HTTP_MESSAGE_RES_BODY:
+        case HTTP_MESSAGE_RES_BODY_END:
+            is_req_msg = 0;
+        break;
+
+        case HTTP_TRANSACTION_NEW:
+        case HTTP_TRANSACTION_FREE:
+        {
+            int cur_dir = packet_get_direction(session_get0_current_packet(sess));
+            if(PACKET_DIRECTION_C2S == cur_dir){
+                is_req_msg = 1;
+            }else{
+                is_req_msg = 0;
+            }
+        }
+        break;
+
+        default:
+            assert(0);
+            fprintf(stderr, "unknow message type:%d\n", (int)msg_type);
+        break;
+    }
+    return is_req_msg;
+}
+
+int http_event_is_req(enum http_event event)
+{
+    switch(event){
+        case HTTP_EVENT_REQ_INIT:
+        case HTTP_EVENT_REQ_LINE:
+        case HTTP_EVENT_REQ_HDR:
+        case HTTP_EVENT_REQ_HDR_END:
+        case HTTP_EVENT_REQ_BODY_BEGIN:
+        case HTTP_EVENT_REQ_BODY_DATA:
+        case HTTP_EVENT_REQ_BODY_END:
+        case HTTP_EVENT_REQ_END:
+            return 1;
+        break;
+
+        case HTTP_EVENT_RES_INIT:
+        case HTTP_EVENT_RES_LINE:
+        case HTTP_EVENT_RES_HDR:
+        case HTTP_EVENT_RES_HDR_END:
+        case HTTP_EVENT_RES_BODY_BEGIN:
+        case HTTP_EVENT_RES_BODY_DATA:
+        case HTTP_EVENT_RES_BODY_END:
+        case HTTP_EVENT_RES_END:
+            return 0;
+        break;
+
+        default:
+            assert(0);
+            fprintf(stderr, "unknow event type:%d\n", (int)event);
+        break;
+    }
+    return -1;
+}
\ No newline at end of file
diff --git a/src/http_decoder_utils.h b/src/http_decoder_utils.h
index 9c031a3..0661641 100644
--- a/src/http_decoder_utils.h
+++ b/src/http_decoder_utils.h
@@ -1,28 +1,13 @@
-/*
-**********************************************************************************************
-*	File: http_decoder_utils.h
-*   Description: 
-*	Authors: LuWenPeng <[email protected]>
-*	Date:    2022-10-31
-*   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);
-
+const char *http_message_type_to_string(enum http_message_type type);
+int http_message_type_is_req(struct session *sess, enum http_message_type msg_type);
+int http_event_is_req(enum http_event event);
 /******************************************************************************
  * Logger
  ******************************************************************************/
@@ -58,9 +43,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