Separate from stellar-on-sapp project.

1 files changed, 858 insertions, 0 deletions

diff --git a/src/http_decoder.c b/src/http_decoder.c
new file mode 100644
index 0000000..68f7c58
--- /dev/null
+++ b/src/http_decoder.c
@@ -0,0 +1,858 @@
+/*
+**********************************************************************************************
+*	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);
+}
+
+int http_message_get_request_header(struct http_message *msg, struct hstring *key,
+	                                struct http_header *hdr_array, size_t array_size)
+{
+	if (NULL == msg || msg->type != HTTP_MESSAGE_REQ_HEADER ||
+        NULL == key || NULL == hdr_array || 0 == array_size) {
+		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_header(req_data, key, hdr_array, array_size);
+}
+
+int http_message_get_response_header(struct http_message *msg, struct hstring *key,
+	                                 struct http_header *hdr_array, size_t array_size)
+{
+	if (NULL == msg || msg->type != HTTP_MESSAGE_RES_HEADER || NULL == key ||
+        NULL == hdr_array || 0 == array_size) {
+		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_header(res_data, key, hdr_array, array_size);
+}
+
+int http_message_request_header_next(struct http_message *msg,
+                                     struct http_header *hdr)
+{
+    if (NULL == msg || msg->type != HTTP_MESSAGE_REQ_HEADER
+        || NULL == hdr) {
+		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_iter_header(req_data, hdr);
+}
+
+int http_message_response_header_next(struct http_message *msg,
+                                      struct http_header *hdr)
+{
+    if (NULL == msg || msg->type != HTTP_MESSAGE_RES_HEADER ||
+        NULL == hdr) {
+		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_iter_header(res_data, hdr);
+}
+
+int http_message_get_request_raw_body(struct http_message *msg,
+                                      struct hstring *body)
+{
+    if (NULL == msg || (msg->type != HTTP_MESSAGE_REQ_BODY) ||
+        NULL == body) {
+        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_raw_body(req_data, body);
+}
+
+int http_message_get_response_raw_body(struct http_message *msg,
+                                       struct hstring *body)
+{
+    if (NULL == msg || (msg->type != HTTP_MESSAGE_RES_BODY) ||
+        NULL == body) {
+        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_raw_body(res_data, body);
+}
+
+int http_message_get_request_decompress_body(struct http_message *msg,
+                                             struct hstring *body)
+{
+    if (NULL == msg || (msg->type != HTTP_MESSAGE_REQ_BODY) ||
+        NULL == body) {
+        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_decompress_body(req_data, body);
+}
+
+int http_message_get_response_decompress_body(struct http_message *msg,
+                                              struct hstring *body)
+{
+    if (NULL == msg || (msg->type != HTTP_MESSAGE_RES_BODY) ||
+        NULL == body) {
+        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_decompress_body(res_data, body);
+}
+
+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);
+}
\ No newline at end of file