path: root/decoders/http/http_decoder_half.c

#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <arpa/inet.h>
#include "http_decoder.h"
#include "http_decoder_half.h"
#include "http_decoder_utils.h"
#include "http_decoder_decompress.h"
#include "llhttp.h"
#include "stellar/session.h"
#include "uthash/utlist.h"
#include "uthash/utarray.h"

#ifdef __cplusplus
extern "C"
{
#endif

    static inline void http_half_stage_reset_flow_data(struct http_half *half)
    {
        if (half->flow_data)
        {
            memset(&half->flow_data->req_line, 0, sizeof(struct http_request_line));
            memset(&half->flow_data->status_line, 0, sizeof(struct http_status_line));
            memset(&half->flow_data->header, 0, sizeof(struct http_header));
            if (half->flow_data->ut_filed_array)
            {
                utarray_clear(half->flow_data->ut_filed_array);
            }
        }
    }
    static inline void http_half_stage_reset_stat(struct http_half *half)
    {
        memset(&half->half_stage, 0, sizeof(struct http_half_llhttp_stage));
    }
    static void http_half_stage_reset(struct http_half *half)
    {
        http_half_stage_reset_stat(half);
        http_half_stage_reset_flow_data(half);
    }

    void http_flow_append_header_filed(struct http_half_data *flow_data, const char *at, size_t length)
    {
        if (NULL == flow_data->ut_filed_array) // the first field
        {
            UT_icd header_icd = {sizeof(struct http_header_field), NULL, NULL, NULL};
            utarray_new(flow_data->ut_filed_array, &header_icd);
            assert(flow_data->ut_filed_array != NULL);
            utarray_reserve(flow_data->ut_filed_array, HTTP_HEADER_FIELD_RESERVE_NUM);
        }
        struct http_header_field new_filed = {};
        new_filed.field_name = (char *)at;
        new_filed.field_name_len = length;
        utarray_push_back(flow_data->ut_filed_array, &new_filed);
    }

    void http_flow_append_header_value(struct http_half_data *flow_data, const char *at, size_t length)
    {
        assert(flow_data->ut_filed_array != NULL);
        struct http_header_field *last_field = (struct http_header_field *)utarray_back(flow_data->ut_filed_array);
        assert(last_field != NULL);
        if (last_field)
        {
            last_field->field_value = (char *)at;
            last_field->field_value_len = length;
        }
    }

    static struct http_message *http_produce_message(struct session *sess, enum http_event event,
                                                     enum http_topic_type topic_type, struct http_half_data *flow_data)
    {
        struct http_message *msg = (struct http_message *)calloc(1, sizeof(struct http_message));
        msg->sess_ref = sess;
        msg->topic_type = topic_type;
        msg->flow_data = flow_data;
        msg->event = event;
        return msg;
    }

    static struct http_half_data *http_half_data_new(enum flow_type flow_dir)
    {
        struct http_half_data *half_data = (struct http_half_data *)calloc(1, sizeof(struct http_half_data));
        half_data->flow_dir = flow_dir;
        /*  update transaction seq in http_half_parse_headers_finally(), not here */
        return half_data;
    }

    void http_half_data_free(struct http_half_data *half_data)
    {
        if (NULL == half_data)
        {
            return;
        }
        if (half_data->ut_filed_array)
        {
            utarray_free(half_data->ut_filed_array);
            half_data->ut_filed_array = NULL;
        }
        if (half_data->joint_url.iov_base)
        {
            FREE(half_data->joint_url.iov_base);
        }
        if (half_data->decompress)
        {
            http_content_decompress_destroy(half_data->decompress);
            half_data->decompress = NULL;
        }
        free(half_data);
    }

    void http_half_free(struct http_half *half)
    {
        if (NULL == half)
        {
            return;
        }
        llhttp_reset(&half->parser.llhttp_parser);
        http_half_data_free(half->flow_data);
        if (half->cached_header_buffer)
        {
            if (half->cached_header_buffer->buffer)
            {
                FREE(half->cached_header_buffer->buffer);
            }
            FREE(half->cached_header_buffer);
        }
        free(half);
    }

    struct http_half *http_half_new(enum flow_type flow_dir)
    {
        struct http_half *half = (struct http_half *)calloc(1, sizeof(struct http_half));
        if (FLOW_TYPE_C2S == flow_dir)
        {
            http_flow_parser_init(&half->parser, HTTP_REQUEST);
        }
        else
        {
            http_flow_parser_init(&half->parser, HTTP_RESPONSE);
        }
        half->parser.half_ref = half;
        return half;
    }

    static struct http_header_field *http_flow_get_header_field(struct http_half_data *flow_data, const char *field_name, size_t field_name_len)
    {
        struct http_header_field *p;
        if (NULL == flow_data->ut_filed_array) // some response no header fileds, such as "HTTP/1.1 100 Continue\r\n\r\n"
        {
            return NULL;
        }
        for (p = (struct http_header_field *)utarray_front(flow_data->ut_filed_array);
             p != NULL;
             p = (struct http_header_field *)utarray_next(flow_data->ut_filed_array, p))
        {
            if ((field_name_len == p->field_name_len) &&
                (strncasecmp(p->field_name, field_name, field_name_len) == 0))
            {
                return p;
            }
        }
        return NULL;
    }

    static void http_using_session_addr_as_host(struct session *ref_session, struct http_header_field *host_result)
    {
        memset(host_result, 0, sizeof(struct http_header_field));
        struct http_session_addr ssaddr = {};
        httpd_session_get_addr(ref_session, &ssaddr);
        char ip_string_buf[INET6_ADDRSTRLEN];

        if (4 == ssaddr.ipver)
        {
            host_result->field_value = (char *)calloc(1, (INET6_ADDRSTRLEN + 7) /* "ip:port" max length */);
            inet_ntop(AF_INET, &ssaddr.daddr4, ip_string_buf, INET6_ADDRSTRLEN);
            snprintf((char *)host_result->field_value, INET6_ADDRSTRLEN + 7, "%s:%u", ip_string_buf, ntohs(ssaddr.dport));
            host_result->field_value_len = strlen(host_result->field_value);
        }
        else if (6 == ssaddr.ipver)
        {
            host_result->field_value = (char *)calloc(1, (INET6_ADDRSTRLEN + 7) /* "ip:port" max length */);
            inet_ntop(AF_INET6, &ssaddr.daddr6, ip_string_buf, INET6_ADDRSTRLEN);
            snprintf((char *)host_result->field_value, INET6_ADDRSTRLEN + 7, "%s:%u", ip_string_buf, ntohs(ssaddr.dport));
            host_result->field_value_len = strlen(host_result->field_value);
        }
        else
        {
            host_result->field_value = (char *)calloc(1, 1);
            host_result->field_value_len = 1;
        }
        return;
    }

    static void http_flow_join_url(struct http_half_data *flow_data, const struct http_header_field *host_filed)
    {
        int append_slash_len = 0;
        const char *join_uri = flow_data->req_line.uri;
        size_t join_url_len = flow_data->req_line.uri_len;

        // skip schema "http://"
        if (join_url_len > 7 && strncasecmp(join_uri, "http://", 7) == 0)
        {
            join_uri += 7;
            join_url_len -= 7;
        }

        // if uri not absolute path(start with '/'), append '/'
        if ('/' != join_uri[0])
        {
            append_slash_len = 1;
        }
        int url_cache_str_len = host_filed->field_value_len + join_url_len + append_slash_len;
        char *url_cache_str = (char *)calloc(1, url_cache_str_len);

        char *ptr = url_cache_str;
        memcpy(ptr, host_filed->field_value, host_filed->field_value_len);
        ptr += host_filed->field_value_len;
        if (append_slash_len)
        {
            *ptr = '/';
            ptr++;
        }
        memcpy(ptr, join_uri, join_url_len);

        flow_data->joint_url.iov_base = url_cache_str;
        flow_data->joint_url.iov_len = url_cache_str_len;
    }

    // todo: optimize use hash table
    static void http_half_get_frequently_used_fileds(struct http_half_data *flow_data)
    {
        if (FLOW_TYPE_C2S == flow_data->flow_dir)
        {
            flow_data->header.host = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_HOST, strlen(HTTP_HEADER_FIELD_NAME_HOST));
            flow_data->header.user_agent = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_USER_AGENT, strlen(HTTP_HEADER_FIELD_NAME_USER_AGENT));
            flow_data->header.referer = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_REFERER, strlen(HTTP_HEADER_FIELD_NAME_REFERER));
            flow_data->header.cookie = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_COOKIE, strlen(HTTP_HEADER_FIELD_NAME_COOKIE));
        }
        else
        {
            flow_data->header.set_cookie = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_SET_COOKIE, strlen(HTTP_HEADER_FIELD_NAME_SET_COOKIE));
        }
        flow_data->header.content_type = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_CONTENT_TYPE, strlen(HTTP_HEADER_FIELD_NAME_CONTENT_TYPE));
    }

    static int http_half_parse_headers_finally(struct http_half *half)
    {
        struct http_half_data *flow_data = half->flow_data;
        http_half_get_frequently_used_fileds(flow_data);

        /* update transaction_seq when headers completed */
        half->flow_data->transaction_seq = half->transaction_num++;

        if (FLOW_TYPE_C2S == flow_data->flow_dir)
        {
            struct http_header_field tmp_host_field = {};
            const struct http_header_field *host_filed = flow_data->header.host;
            if (NULL == host_filed)
            {
                http_using_session_addr_as_host(half->sess_ref, &tmp_host_field);
                host_filed = &tmp_host_field;
            }
            http_flow_join_url(flow_data, host_filed);
            FREE(tmp_host_field.field_name);
            FREE(tmp_host_field.field_value);
        }

        struct http_header_field *encoding_field = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_CONTENT_ENCODING, strlen(HTTP_HEADER_FIELD_NAME_CONTENT_ENCODING));
        if (NULL == encoding_field)
        {
            flow_data->content_encoding_type = HTTP_CONTENT_ENCODING_NONE;
        }
        else
        {
            flow_data->content_encoding_type = http_content_encoding_str2int(encoding_field->field_value, encoding_field->field_value_len);
        }

        struct http_header_field *content_len_field = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_CONTENT_LENGTH, strlen(HTTP_HEADER_FIELD_NAME_CONTENT_LENGTH));
        if (NULL == content_len_field)
        {
            flow_data->header.content_length = -1;
        }
        else
        {
            flow_data->header.content_length = http_strtoll(content_len_field->field_value, content_len_field->field_value_len);
        }

        struct http_header_field *transf_encoding_field = http_flow_get_header_field(flow_data, HTTP_HEADER_FIELD_NAME_TRANSFER_ENCODING, strlen(HTTP_HEADER_FIELD_NAME_TRANSFER_ENCODING));
        if (NULL == transf_encoding_field)
        {
            flow_data->transfer_encoding_is_chunked = -1;
        }
        else
        {
            if (http_strncasecmp_safe(transf_encoding_field->field_value, "chunked", transf_encoding_field->field_value_len, strlen("chunked")) == 0)
            {
                flow_data->transfer_encoding_is_chunked = 1;
            }
            else
            {
                flow_data->transfer_encoding_is_chunked = 0;
            }
        }
        flow_data->header.header_buf = half->half_stage.first_header_name_ptr;
        flow_data->header.header_buf_sz = half->half_stage.last_headers_complete_ptr - half->half_stage.first_header_name_ptr;

        // todo, parse other frequently used fields
        return 0;
    }

    int http_get_header_field_count(struct http_half_data *flow_data)
    {
        if (NULL == flow_data->ut_filed_array)
        {
            return 0; // some response no header fileds, such as "HTTP/1.1 100 Continue\r\n\r\n"
        }
        return utarray_len(flow_data->ut_filed_array);
    }

    void http_flow_body_decompress(struct http_half_data *flow_data, const char *zipdata, size_t zipdatalen)
    {
        assert(flow_data);
        if (flow_data->content_encoding_type == HTTP_CONTENT_ENCODING_NONE)
        {
            return;
        }
        if (zipdata == NULL || zipdatalen == 0)
        {
            return;
        }
        if (NULL == flow_data->decompress)
        {
            flow_data->decompress = http_content_decompress_create(flow_data->content_encoding_type);
        }

        char *local_outdata = NULL;
        size_t local_outdata_len = 0;
        if (http_content_decompress_write(flow_data->decompress, (char *)zipdata,
                                          zipdatalen, &local_outdata, &local_outdata_len) == -1)
        {
            http_content_decompress_destroy(flow_data->decompress);
            flow_data->decompress = NULL;
            flow_data->decompress_body.body = NULL;
            flow_data->decompress_body.body_sz = 0;
            return;
        }

        if (local_outdata != NULL && local_outdata_len > 0)
        {
            flow_data->decompress_body.body = local_outdata;
            flow_data->decompress_body.body_sz = local_outdata_len;
            http_content_decompress_ownership_borrow(flow_data->decompress);
        }
    }

    static void http_add_header_buf_reference(struct http_half *half)
    {
        if (half->cached_header_buffer != NULL)
        {
            // add reference but not move ownership, because the buffer will be referenced of multiple transaction in pipeline mode
            half->cached_header_buffer->reference++;
            half->flow_data->cached_header_buffer = half->cached_header_buffer;
        }
    }

    static void http_update_body_offset(struct http_half_data *flow_data)
    {
        if (flow_data->decompress_body.body != NULL || flow_data->decompress_body.offset > 0)
        {
            flow_data->decompress_body.offset += flow_data->decompress_body.body_sz;
        }
        else
        {
            flow_data->raw_body.offset += flow_data->raw_body.body_sz;
        }
    }

    static void http_set_body_finish(struct http_half_data *flow_data)
    {
        if (flow_data->decompress_body.offset > 0)
        {
            flow_data->decompress_body.body = NULL;
            flow_data->decompress_body.body_sz = 0;
            flow_data->decompress_body.is_finished = 1;
        }
        else
        {
            flow_data->raw_body.body = NULL;
            flow_data->raw_body.body_sz = 0;
            flow_data->raw_body.is_finished = 1;
        }
    }

    void http_event_handler(struct http_half *half, enum http_event event, struct http *http_env)
    {
        struct http_message *msg = NULL;
        int thread_id = module_manager_get_thread_id(http_env->mod_mgr_ref);
        uint8_t flow_flag = 0;
        struct mq_runtime *mq_rt = module_manager_get_mq_runtime(http_env->mod_mgr_ref);

        switch (event)
        {
        case HTTP_EVENT_REQ_INIT:
            http_half_stage_reset(half);
            if (NULL == half->flow_data) /* call llhttp_reset when headers is not completed */
            {
                half->flow_data = http_half_data_new(FLOW_TYPE_C2S);
            }
            break;

        case HTTP_EVENT_REQ_HDR_END:
        {
            http_half_parse_headers_finally(half);
            int tot_c2s_headers = http_get_header_field_count(half->flow_data);
            http_stat_update(&http_env->stat, thread_id, HTTP_TRANSACTION_NEW, 1);
            http_stat_update(&http_env->stat, thread_id, HTTP_C2S_HEADERS, tot_c2s_headers);
            http_stat_update(&http_env->stat, thread_id, HTTP_URL_BYTES, half->flow_data->joint_url.iov_len);
            http_add_header_buf_reference(half);
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_REQ_HEADER, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_REQ_HEADER].topic_id, msg);
        }
        break;
        case HTTP_EVENT_REQ_BODY_BEGIN:
            break;
        case HTTP_EVENT_REQ_BODY_DATA:
        {
            http_update_body_offset(half->flow_data);
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_REQ_BODY, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_REQ_BODY].topic_id, msg);
        }
        break;
        case HTTP_EVENT_REQ_BODY_END:
        {
            http_set_body_finish(half->flow_data);
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_REQ_BODY, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_REQ_BODY].topic_id, msg);
        }
        break;
        case HTTP_EVENT_REQ_END:
        {
            if ((0 == session_is_symmetric(half->sess_ref, &flow_flag) && (SESSION_SEEN_C2S_FLOW == flow_flag)))
            {
                http_stat_update(&http_env->stat, thread_id, HTTP_TRANSACTION_FREE, 1);
                http_stat_update(&http_env->stat, thread_id, HTTP_C2S_ASYMMETRY_TRANSACTION, 1);
            }
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_REQ_BODY, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_REQ_BODY].topic_id, msg);
            half->flow_data = NULL; // ownership move to message, free it in message free callback
        }
        break;

        case HTTP_EVENT_RES_INIT:
        {
            http_half_stage_reset(half);
            if (NULL == half->flow_data) /* call llhttp_reset when headers is not completed */
            {
                half->flow_data = http_half_data_new(FLOW_TYPE_S2C);
            }
        }
        break;

        case HTTP_EVENT_RES_HDR_END:
        {
            http_half_parse_headers_finally(half);
            int tot_s2c_headers = http_get_header_field_count(half->flow_data);
            http_stat_update(&http_env->stat, thread_id, HTTP_S2C_HEADERS, tot_s2c_headers);
            if ((0 == session_is_symmetric(half->sess_ref, &flow_flag) && (SESSION_SEEN_S2C_FLOW == flow_flag)))
            {
                http_stat_update(&http_env->stat, thread_id, HTTP_TRANSACTION_NEW, 1);
            }
            http_add_header_buf_reference(half);
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_RES_HEADER, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_RES_HEADER].topic_id, msg);
        }
        break;
        case HTTP_EVENT_RES_BODY_BEGIN:
            break;
        case HTTP_EVENT_RES_BODY_DATA:
        {
            http_update_body_offset(half->flow_data);
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_RES_BODY, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_RES_BODY].topic_id, msg);
        }
        break;
        case HTTP_EVENT_RES_BODY_END:
        {
            http_set_body_finish(half->flow_data);
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_RES_BODY, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_RES_BODY].topic_id, msg);
        }
        break;
        case HTTP_EVENT_RES_END:
        {
            if ((0 == session_is_symmetric(half->sess_ref, &flow_flag) && (SESSION_SEEN_S2C_FLOW == flow_flag)))
            {
                http_stat_update(&http_env->stat, thread_id, HTTP_S2C_ASYMMETRY_TRANSACTION, 1);
            }
            http_stat_update(&http_env->stat, thread_id, HTTP_TRANSACTION_FREE, 1);
            msg = http_produce_message(half->sess_ref, event, HTTP_TOPIC_RES_BODY, half->flow_data);
            mq_runtime_publish_message(mq_rt, http_env->http_topic_mgr->topic_compose[HTTP_TOPIC_RES_BODY].topic_id, msg);
            half->flow_data = NULL; // ownership move to message, free it in message free callback
        }
        break;
        default:
            assert(0);
            break;
        }
    }

    static void http_half_stage_buf_free(struct http_half *half)
    {
        if (half->cached_header_buffer != NULL)
        {
            http_buffer_free(half->cached_header_buffer);
        }
        half->cached_header_buffer = NULL;
    }

    static void http_half_stage_buf_reuse(struct http_half *half)
    {
        http_half_stage_buf_free(half);
        half->cached_header_buffer = http_buffer_new();
    }

    static int http_half_stage_buf_append(struct http_half *half, const char *newdata, size_t newdata_len)
    {
        if (half->cached_header_buffer == NULL)
        {
            half->cached_header_buffer = http_buffer_new();
        }
        if (half->cached_header_buffer->buffer_size > HTTP_HEADERS_CACHE_MAX_SIZE)
        {
            return -1;
        }
        http_buffer_add(half->cached_header_buffer, newdata, newdata_len);
        return 0;
    }

    static int http_half_cache_uncompleted_header(struct http_half *half, int mem_merged, const char *data, size_t data_len)
    {
        int ret = 0;
        const char *cached_begin_ptr = NULL;
        long long cached_len = 0;
        if (half->half_stage.last_headers_complete_ptr == NULL)
        {
            if (mem_merged)
            {
                ret = 0; // no pipeline and memory have been merged, do nothing
            }
            else
            {
                ret = http_half_stage_buf_append(half, data, (size_t)data_len);
            }
        }
        else
        {
            /*
                has pipeline, cache begin ptr is the last completed header_field_value + "\r\n\r\n",
                for example:
                GET xxx \r\nheader_name:header:value\r\nheader_name:header:value\r\n\r\nGET yyy ...
                                                                                        ^
                                                                                        |
                                                                                        cache begin ptr
            */
            cached_begin_ptr = half->half_stage.last_headers_complete_ptr;
            cached_len = (long long)data_len - (long long)(cached_begin_ptr - data);
            assert(cached_len > 0);

            if (mem_merged)
            {
                /* some message have beed pushed, and pointer to half->half_stage->half_buffer,
                   the half->half_stage->half_buffer is maintain by messages ,
                   so, we can reset the half->half_stage->half_buffer, and append the new data to it.
                */
                http_half_stage_buf_reuse(half);
                ret = http_half_stage_buf_append(half, cached_begin_ptr, (size_t)cached_len);
            }
            else
            {
                ret = http_half_stage_buf_append(half, cached_begin_ptr, (size_t)cached_len);
            }
        }
        return ret;
    }

    int http_flow_parse(struct http_half *half, const char *data, size_t data_len)
    {
        assert(half && data);
        llhttp_errno_t ret = HPE_OK /* HPE_OK = 0 */;
        ret = llhttp_execute(&half->parser.llhttp_parser, data, data_len);
        switch (ret)
        {
        case HPE_OK:
            break;
        case HPE_PAUSED:
            llhttp_resume(&half->parser.llhttp_parser);
            break;
        case HPE_PAUSED_UPGRADE:
            llhttp_resume_after_upgrade(&half->parser.llhttp_parser);
            break;
        default:
            break;
        }
        return ret;
    }

    struct http_half *http_flow_get_nx(struct http_exdata *exdata, enum flow_type flow_dir, struct session *sess, struct http *http_env)
    {
        struct http_half **flow = NULL;

        if (FLOW_TYPE_C2S == flow_dir)
        {
            flow = &exdata->decoder->flow_c2s;
        }
        else if (FLOW_TYPE_S2C == flow_dir)
        {
            flow = &exdata->decoder->flow_s2c;
        }
        else
        {
            assert(0);
            return NULL;
        }
        if (NULL == *flow)
        {
            *flow = http_half_new(flow_dir);
            (*flow)->http_env_ref = http_env;
            (*flow)->sess_ref = sess;
        }
        return *flow;
    }

    int http_half_flow_process(struct http_half *half, const char *newdata, size_t newdata_len)
    {
        int ret = 0;
        int mem_merged = 0;
        const char *parse_data = newdata;
        size_t parse_data_len = newdata_len;

        /* merge cached uncompleted header and current tcp payload first */
        if (half->cached_header_buffer != NULL && half->cached_header_buffer->buffer != NULL)
        {
            http_half_stage_buf_append(half, newdata, newdata_len);
            parse_data = half->cached_header_buffer->buffer;
            parse_data_len = half->cached_header_buffer->buffer_size;
            mem_merged = 1;
            if (parse_data_len > HTTP_HEADERS_CACHE_MAX_SIZE)
            {
                STELLAR_LOG_FATAL(half->http_env_ref->logger_ref, HTTP_MODULE_NAME,
                                  "sess: %s, headers buf len more than %d", session_get_readable_addr(half->sess_ref), HTTP_HEADERS_CACHE_MAX_SIZE);
                return -1;
            }
        }
        ret = http_flow_parse(half, parse_data, parse_data_len);
        if (ret != HPE_OK)
        {
            STELLAR_LOG_FATAL(half->http_env_ref->logger_ref, HTTP_MODULE_NAME,
                              "llhttp parser error: %d, %s. sess: %s", ret, llhttp_errno_name(ret), session_get_readable_addr(half->sess_ref));
            return -1;
        }

        if (half->half_stage.llhttp_last_stage >= LLHTTP_STAGE_MESSAGE_BEGIN &&
            half->half_stage.llhttp_last_stage < LLHTTP_STAGE_HEADERS_COMPLETE) /* some headers not completed */
        {
            STELLAR_LOG_DEBUG(half->http_env_ref->logger_ref, HTTP_MODULE_NAME,
                              "sess: %s, header not completed, need caching, %.*s",
                              session_get_readable_addr(half->sess_ref), parse_data_len > 16 ? 16 : parse_data_len, parse_data);
            if (http_half_cache_uncompleted_header(half, mem_merged, parse_data, parse_data_len) < 0)
            {
                STELLAR_LOG_FATAL(half->http_env_ref->logger_ref, HTTP_MODULE_NAME,
                                  "sess: %s, headers buf len more than %d", session_get_readable_addr(half->sess_ref), HTTP_HEADERS_CACHE_MAX_SIZE);
                return -1;
            }

            /* uncompleted_header, reset llhttp state and all headers */
            http_half_stage_reset(half);
            llhttp_reset(&half->parser.llhttp_parser);
            return 0;
        }
        else
        {
            /* ownership is maintained by header message, free it here (reference-- actually) */
            http_half_stage_buf_free(half);
        }
        return 0;
    }

#ifdef __cplusplus
}
#endif