path: root/src/http1.cc

/* HTTP/HTTP2 Protocol Handler
 *
 * Author: Lu Qiuwen<[email protected]>
 * Date: 2018-04-08
 *
 */

#include <vector>
#include <array>
#include <list>
#include <map>
#include <memory>
#include <cassert>

#include <http_parser.h>
#include <event2/bufferevent.h>
#include <evdns.h>

#include "pxyconn.h"
#include "http.h"
#include "httpscan.h"
#include "compat.h"
#include "util.h"
#include "easylogging++.h"

class Http1RequestParserCallbacks;
class Http1Request;
class Http1Response;

class Http1Headers : public HttpHeaders
{
public:
    void Add(const str_field_t & str_field, const str_value_t & str_value) override
    {
        headers_store_.emplace_back(str_field, str_value);
    }

    void Set(const str_field_t & str_field, const str_value_t & str_value) override
    {
        Remove(str_field);
        Add(str_field, str_value);
    }

    void Remove(const str_field_t & str_field) override
    {
        headers_store_.erase(std::remove_if(headers_store_.begin(), headers_store_.end(),
            [&str_field](kv_pair_t & kv) { return strcasecmp(str_field.c_str(), kv.first.c_str()) == 0; }),
            headers_store_.end());
    }

    bool ForEachHeader(for_each_cb_t cb) const override
    {
        bool is_callbacked = false;

        for (const kv_pair_t & kv_pair : headers_store_)
        {
            is_callbacked = true;
            bool cb_ret = cb(kv_pair.first, kv_pair.second);
            if (!cb_ret) break;
        }

        return is_callbacked;
    }

    bool ForEachValueOfHeader(const str_field_t & str_field, for_each_cb_t cb) const override
    {
        bool is_callbacked = false;

        for (const kv_pair_t & kv_pair : headers_store_)
        {
            if (strcasecmp(str_field.c_str(), kv_pair.first.c_str()) != 0)
                continue;

            is_callbacked = true;
            bool cb_ret = cb(kv_pair.first, kv_pair.second);
            if (!cb_ret) break;
        }

        return is_callbacked;
    }

    evbuffer_unique_ptr_t SerializeToEvBuffer() const
    {
        auto __evbuffer_ptr = evbuffer_unique_ptr_t(evbuffer_new());

        for (const auto & __iterate : headers_store_)
        {
            const auto & __header_field = __iterate.first;
            const auto & __header_value = __iterate.second;

            evbuffer_add_printf(__evbuffer_ptr.get(), "%s: %s\r\n",
                __header_field.c_str(), __header_value.c_str());
        }

        return std::move(__evbuffer_ptr);
    }

    void ConstructByHttpParserField(const char * data, size_t len)
    {
        if (!str_last_value_field_.empty())
            ConstructByHttpParserComplete();

        /* 新的K-V对或Continued Field， Value不应该有值 */
        assert(str_last_value_field_.empty());
        str_last_header_field_ += std::string(data, data + len);
    }

    void ConstructByHttpParserValue(const char * data, size_t len)
    {
        /* 处理到Value时，Field不应为空
         * Value可能分成片段，多次调用给出，在这里需要将他们拼合 */

        /* Hack, 直接覆盖Accept-Encoding，用空格等长替换 */
        if (strcasecmp(str_last_header_field_.c_str(), "Accept-Encoding") == 0)
        {
            memset((void *)data, 0x20, len);
        }

        str_last_value_field_ += std::string(data, data + len);
    }

    void ConstructByHttpParserComplete()
    {
        /* 收到了一个完成的Header K-V，写入Header字段 */
        if (!str_last_header_field_.empty())
        {
            Add(str_last_header_field_, str_last_value_field_);
        }
        else
        {
            LOG(WARNING) << "Empty HTTP Header field recieved, drop this header-value.";
        }

        str_last_header_field_.clear();
        str_last_value_field_.clear();

        /* 完成时，不应该有值没有写入Header记录 */
        assert(str_last_header_field_.empty());
        assert(str_last_value_field_.empty());
    }

private:
    /* Type Declare */
    using kv_pair_t = std::pair<str_field_t, str_value_t>;
    using kv_store_t = std::vector<kv_pair_t>;

    /* Store */
    kv_store_t headers_store_{};

    /* Construct: HttpParser */
    std::string str_last_header_field_;
    std::string str_last_value_field_;
};

class Http1Request : public HttpRequest
{
public:
    explicit Http1Request(short major_version, short minor_version);
    virtual ~Http1Request() = default;

    virtual ssize_t ConstructFromMemory(const char * buf, size_t buflen);
    virtual ssize_t ConstructFromEvBuf(struct evbuffer * evbuf_ptr);

    virtual evbuffer_unique_ptr_t StolenEvbuf()
    { return std::move(evbuf_content_raw_); }

    bool ReadOnly() override
    { return readonly_; }
    void ReadOnly(bool is_readonly) override
    { readonly_ = is_readonly; }

    bool Forward() override
    { return forward_; }
    void Forward(bool is_forward) override
    { forward_ = is_forward; }

    bool Complete(section_t section) override;

    const std::string & Url() const override
    { return str_url_; }
    void Url(const std::string & url) override
    { assert(0); }

    const body_content_t * Body() const override
    { return body_content_.get(); }
    void Body(body_content_ptr_t body) override
    { body_content_ = std::move(body); }
    body_content_ptr_t StolenBody() override
    { return std::move(body_content_); }

    version_t Version() override
    { return version_t{major_version_, minor_version_}; }

    void Construct() override;

    HttpHeaders & Headers() override
    { return headers_; }

    const HttpHeaders & cHeaders() const override
    { return headers_; }

    const std::string & Uri() const override
    { return str_uri_; }

    void Uri(const std::string & url) override
    { str_uri_ = url; }

    bool Bypass() override
    { return false; }
    void Bypass(bool is_bypass) override
    {}

    std::string DumpToString() override;

private:
    /* Http Version */
    short major_version_{1};
    short minor_version_{0};
    unsigned int method_{0};

    /* Url */
    std::string str_uri_{};
    std::string str_url_{};

    /* Headers */
    Http1Headers headers_{};

    /* Last Header */
    bool is_last_header_kv_set{false};
    std::string str_last_header_field_{};
    std::string str_last_value_field_{};

    /* Pointer to Request Body */
    body_content_ptr_t body_content_{nullptr};

    /* Tag */
    bool request_header_complete_{false};
    bool request_body_complete_{false};

    bool forward_{true};
    bool readonly_{false};

    /* Underlay Buffer */
    std::unique_ptr<struct http_parser> parser_{new struct http_parser};
    evbuffer_unique_ptr_t evbuf_content_raw_{evbuffer_unique_ptr_t(evbuffer_new())};

    /* Allow callback handlers to access internal data */
    friend Http1RequestParserCallbacks;
};

class Http1RequestParserCallbacks
{
public:
    static const http_parser_settings * CallbackSettings();
    static int CallbackOnUri(http_parser * parser, const char * at, size_t length);
    static int CallbackOnHeaderField(http_parser * parser, const char * at, size_t length);
    static int CallbackOnHeaderValue(http_parser * parser, const char * at, size_t length);
    static int CallbackOnBody(http_parser * parser, const char * at, size_t length);
    static int CallbackOnHeaderComplete(http_parser * parser);
    static int CallbackOnMessageComplete(http_parser * parser);

private:
    static Http1Request * __request_this_ptr(http_parser * parser)
    { return static_cast<Http1Request *>(parser->data); }
};

const http_parser_settings * Http1RequestParserCallbacks::CallbackSettings()
{
    static struct http_parser_settings __parser_setting =
        {
            on_message_begin : nullptr, /* on_message_begin */
            on_url : CallbackOnUri, /* on_url */
            on_status : nullptr, /* on_status */
            on_header_field : CallbackOnHeaderField, /* on_header_field */
            on_header_value : CallbackOnHeaderValue, /* on_header_value */
            on_headers_complete : CallbackOnHeaderComplete, /* on_headers_complete */
            on_body : CallbackOnBody, /* on_body */
            on_message_complete : CallbackOnMessageComplete /* on_message_complete */
        };

    return &__parser_setting;
}

int Http1RequestParserCallbacks::CallbackOnUri(http_parser * parser, const char * at, size_t length)
{
    __request_this_ptr(parser)->str_uri_ += std::string(at, at + length);
    return 0;
}

int Http1RequestParserCallbacks::CallbackOnHeaderField(http_parser * parser, const char * at, size_t length)
{
    __request_this_ptr(parser)->headers_.ConstructByHttpParserField(at, length);
    return 0;
}

int Http1RequestParserCallbacks::CallbackOnHeaderValue(http_parser * parser, const char * at, size_t length)
{
    __request_this_ptr(parser)->headers_.ConstructByHttpParserValue(at, length);
    return 0;
}

int Http1RequestParserCallbacks::CallbackOnBody(http_parser * parser, const char * at, size_t length)
{
    auto * request_ptr = __request_this_ptr(parser);

    /* 第一次回调Body，分配内存 */
    if (request_ptr->body_content_ == nullptr)
    {
        request_ptr->body_content_ = std::make_unique<std::vector<char>>();
    }

    /* 增加length长度的缓冲区，并在尾部追加 */
    request_ptr->body_content_->insert(request_ptr->body_content_->end(), at, at + length);
    return 0;
}

int Http1RequestParserCallbacks::CallbackOnHeaderComplete(http_parser * parser)
{
    auto * __request = __request_this_ptr(parser);

    /* HTTP版本号、请求类型 */
    __request->major_version_ = parser->http_major;
    __request->minor_version_ = parser->http_minor;
    __request->method_ = parser->method;

    /* 最后一个头部K-V */
    __request->headers_.ConstructByHttpParserComplete();

    /* 拼合URL，TODO: 处理拼合过程中的各种异常，例如多一个斜线、少一个斜线等问题 */
    __request->headers_.ForEachValueOfHeader("Host",
        [__request](const std::string & field, const std::string & value)
        {
            __request->str_url_ = value + __request->str_uri_;
            return false;
        });

    __request->request_header_complete_ = true;
    return 0;
}

int Http1RequestParserCallbacks::CallbackOnMessageComplete(http_parser * parser)
{
    __request_this_ptr(parser)->request_body_complete_ = true;
    return 0;
}

Http1Request::Http1Request(short major_version, short minor_version) :
    major_version_(major_version), minor_version_(minor_version)
{
    http_parser_init(parser_.get(), HTTP_REQUEST);
}

ssize_t Http1Request::ConstructFromMemory(const char * buf, size_t buflen)
{
    parser_->data = this;

    size_t sz_parsed = http_parser_execute(parser_.get(),
        Http1RequestParserCallbacks::CallbackSettings(), buf, buflen);

    /* 解析失败 */
    if(sz_parsed && parser_->http_errno > 0)
    {
        throw std::runtime_error(string_format("Failed at http parsing: errcode=%u, %s, %s",
            parser_->http_errno, http_errno_name(static_cast<http_errno>(parser_->http_errno)),
            http_errno_description(static_cast<http_errno>(parser_->http_errno))));
    }

    return sz_parsed;
}

ssize_t Http1Request::ConstructFromEvBuf(struct evbuffer * evbuf_ptr)
{
    /* 展平输入的Buffer为线性空间 */
    auto * __data_ptr = (const char *) (evbuffer_pullup(evbuf_ptr, -1));
    size_t __data_len = evbuffer_get_length(evbuf_ptr);

    /* 解析输入的Buffer */
    ssize_t forward_len = ConstructFromMemory(__data_ptr, __data_len);
    if (forward_len < 0) return forward_len;

    /* 已解析的部分，从原Buffer中抽离出来 */
    struct evbuffer * reserved_buffer = evbuffer_new();
    evbuffer_remove_buffer(evbuf_ptr, reserved_buffer, static_cast<size_t>(forward_len));

    /* 保存在上下文中 */
    evbuffer_add_buffer(evbuf_content_raw_.get(), reserved_buffer);
    evbuffer_free(reserved_buffer);

    return forward_len;
}

bool Http1Request::Complete(section_t section)
{
    if (section == kSectionHeader) return request_header_complete_;
    if (section == kSectionBody) return request_body_complete_;
    if (section == kSecionMessage) return request_body_complete_ && request_header_complete_;
}

void Http1Request::Construct()
{
    /* 新申请EvBuffer */
    evbuffer_unique_ptr_t evbuf_construct = evbuffer_unique_ptr_t(evbuffer_new());
    struct evbuffer * evbuf_output_ptr = evbuf_construct.get();

    /* Method序号转字符串，不认识的序号会转换成<unknown>
     * TODO: 增加<unknown>的告警 */
    const char * str_method = http_method_str(static_cast<enum http_method>(method_));

    /* 请求第一行 */
    evbuffer_add_printf(evbuf_output_ptr, "%s %s HTTP/%d.%d\r\n", str_method,
        str_uri_.c_str(), major_version_, minor_version_);

    /* 请求头部 */
    auto headers_evbuffer_ptr = headers_.SerializeToEvBuffer();
    evbuffer_add_buffer(evbuf_output_ptr, headers_evbuffer_ptr.release());

    /* 请求头部与请求体的分隔符 */
    evbuffer_add_printf(evbuf_output_ptr, "\r\n");
    evbuffer_add(evbuf_output_ptr, body_content_->data(), body_content_->size());

    /* 替换原有的请求 */
    evbuf_content_raw_ = std::move(evbuf_construct);
    return;
}

std::string Http1Request::DumpToString()
{
    return string_format("HTTP/%d.%d %s", major_version_, minor_version_, str_url_.c_str());
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//  HTTP Response Parser Implementation for HTTP 1.0/1.1
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

static const char * informational_phrases[] = {
    /* 100 */ "Continue",
    /* 101 */ "Switching Protocols"
};

static const char * success_phrases[] = {
    /* 200 */ "OK",
    /* 201 */ "Created",
    /* 202 */ "Accepted",
    /* 203 */ "Non-Authoritative Information",
    /* 204 */ "No Content",
    /* 205 */ "Reset Content",
    /* 206 */ "Partial Content"
};

static const char * redirection_phrases[] = {
    /* 300 */ "Multiple Choices",
    /* 301 */ "Moved Permanently",
    /* 302 */ "Found",
    /* 303 */ "See Other",
    /* 304 */ "Not Modified",
    /* 305 */ "Use Proxy",
    /* 307 */ "Temporary Redirect"
};

static const char * client_error_phrases[] = {
    /* 400 */ "Bad Request",
    /* 401 */ "Unauthorized",
    /* 402 */ "Payment Required",
    /* 403 */ "Forbidden",
    /* 404 */ "Not Found",
    /* 405 */ "Method Not Allowed",
    /* 406 */ "Not Acceptable",
    /* 407 */ "Proxy Authentication Required",
    /* 408 */ "Request Time-out",
    /* 409 */ "Conflict",
    /* 410 */ "Gone",
    /* 411 */ "Length Required",
    /* 412 */ "Precondition Failed",
    /* 413 */ "Request Entity Too Large",
    /* 414 */ "Request-URI Too Large",
    /* 415 */ "Unsupported Media Type",
    /* 416 */ "Requested range not satisfiable",
    /* 417 */ "Expectation Failed",
    /* 418 */ nullptr,
    /* 419 */ nullptr,
    /* 420 */ nullptr,
    /* 421 */ "Misdirected Request",
    /* 422 */ "Unprocessable Entity",
    /* 423 */ "Locked",
    /* 424 */ "Failed Dependency",
    /* 425 */ nullptr,
    /* 426 */ "Upgrade Required",
    /* 427 */ nullptr,
    /* 428 */ "Precondition Required",
    /* 429 */ "Too Many Requests"
};

static const char * client_error_phrases_43x[] = {
    /* 430 */ nullptr,
    /* 431 */ "Request Header Fields Too Large"
};

static const char * client_error_phrases_45x[] = {
    /* 450 */ nullptr,
    /* 451 */ "Unavailable For Legal Reasons"
};

static const char * server_error_phrases[] = {
    /* 500 */ "Internal Server Error",
    /* 501 */ "Not Implemented",
    /* 502 */ "Bad Gateway",
    /* 503 */ "Service Unavailable",
    /* 504 */ "Gateway Time-out",
    /* 505 */ "HTTP Version not supported"
};

const char * __resp_code_to_str(int resp_code)
{
    if (resp_code >= 101 && resp_code <= 102) return informational_phrases[resp_code - 101];
    if (resp_code >= 200 && resp_code <= 206) return success_phrases[resp_code - 200];
    if (resp_code >= 300 && resp_code <= 307) return redirection_phrases[resp_code - 300];
    if (resp_code >= 400 && resp_code <= 429) return client_error_phrases[resp_code - 400];
    if (resp_code >= 430 && resp_code <= 431) return client_error_phrases_43x[resp_code - 430];
    if (resp_code >= 450 && resp_code <= 451) return client_error_phrases_45x[resp_code - 450];
    if (resp_code >= 500 && resp_code <= 505) return server_error_phrases[resp_code - 500];

    return nullptr;
}

class Http1ResponseParserCallbacks;
class Http1Response : public HttpResponse
{
public:
    Http1Response(short major_version, short minor_version);
    virtual ~Http1Response() = default;

    virtual ssize_t ConstructFromMemory(const char * buf, size_t buflen);
    virtual ssize_t ConstructFromEvBuf(struct evbuffer * evbuf_ptr);

    evbuffer_unique_ptr_t StolenEvBuf();

    int ResponseCode() override
    { return resp_code_; }
    void ResponseCode(int code) override
    { resp_code_ = code; }

    bool ReadOnly() override
    { return readonly_; }
    void ReadOnly(bool is_readonly) override
    { readonly_ = is_readonly; }

    bool Forward() override
    { return forward_; }
    void Forward(bool is_forward) override
    { forward_ = is_forward; }

    bool Bypass() override
    { return bypass_; }

    void Bypass(bool is_bypass) override
    { bypass_ = true; }

    section_state_t SectionState(section_t section) override
    { return section_state[section]; }

    void Construct() override;

    HttpHeaders & Headers() override
    { return headers_; }
    const HttpHeaders & cHeaders() const override
    { return headers_; }

    const std::vector<const body_content_t *> Body() const override
    {
        std::vector<const body_content_t *> body_const;
        for (auto & body_content_iter : body_contents_)
        {
            if (!body_content_iter.is_complete) continue;
            body_const.push_back(body_content_iter.ptr.get());
        }

        return body_const;
    }

    std::vector<body_content_ptr_t> StolenBody() override
    {
        std::vector<body_content_ptr_t> body_ret;
        for (auto & body_content_iter : body_contents_)
        {
            body_ret.push_back(std::move(body_content_iter.ptr));
        }

        body_contents_.clear();
        return body_ret;
    }

    void Body(std::vector<body_content_ptr_t> body) override
    {
        assert(body_contents_.empty());
        body_contents_.clear();

        for (auto & body_content_iter : body)
        {
            body_contents_.emplace_back(std::move(body_content_iter), true);
        }
    }

    std::string DumpToString() override;

private:
    int resp_code_{-1};
    short resp_version_major_{1};
    short resp_version_minor_{1};

    bool forward_{true};
    bool readonly_{false};
    bool bypass_{false};

    /* Tags */
    section_state_t section_state[kSectionMax]{kStateBegin};

    /* Construct Tags */
    bool construct_headers_{true};
    bool construct_body_{true};

    /* Body */
    struct __body_content
    {
        __body_content(body_content_ptr_t __ptr, bool __is_complete)
            : ptr(std::move(__ptr)), is_complete(__is_complete) {}

        body_content_ptr_t ptr{nullptr};
        bool is_complete{false};
    };

    std::vector<struct __body_content> body_contents_;

    /* Chunk */
    bool body_is_chunk{false};
    unsigned int body_chunk_id{0};

    Http1Headers body_chunk_headers{};
    std::string body_chunk_last_field_;

    /* Headers */
    Http1Headers headers_{};
    std::string last_header_field_{};

    /* Headers For Debug */
    std::string content_type{};
    std::string content_length{};
    std::string content_encoding{};

    /* Parser & Raw Content
     * 该原始内容，等价于本结构体的序列化后的结果，因此，输入的、未经解析的字节数据，不应该保留在这个位置 */
    std::unique_ptr<struct http_parser> parser_{new struct http_parser};
    evbuffer_unique_ptr_t evbuf_content_raw_{evbuffer_unique_ptr_t(evbuffer_new())};

    /* Friend to Callbacks */
    friend Http1ResponseParserCallbacks;

private:
    void __set_section_state(section_t section, section_state_t state)
    { section_state[section] = state; }

    void __new_body_content(size_t content_length)
    {
        body_contents_.emplace_back(std::make_unique<body_content_t>(), false);
        if (content_length > 0) __last_body_content().ptr->reserve(content_length);
    }

    struct __body_content & __last_body_content()
    { return body_contents_.back(); }

    bool __is_body_content_empty()
    { return body_contents_.empty(); }

    void __drop_all_body_content()
    { body_contents_.clear(); }
};

class Http1ResponseParserCallbacks
{
public:
    static http_parser_settings * CallbackSettings();
    static int CallbackOnMessageBegin(http_parser * parser);
    static int CallbackOnHeaderField(http_parser * parser, const char * at, size_t length);
    static int CallbackOnHeaderValue(http_parser * parser, const char * at, size_t length);
    static int CallbackOnBody(http_parser * parser, const char * at, size_t length);
    static int CallbackOnHeaderComplete(http_parser * parser);
    static int CallbackOnMessageComplete(http_parser * parser);
    static int CallbackOnChunkHeader(http_parser * parser);
    static int CallbackOnChunkComplete(http_parser * parser);

private:
    static Http1Response * __response_this_ptr(http_parser * parser)
    { return static_cast<Http1Response *>(parser->data); }
};

http_parser_settings * Http1ResponseParserCallbacks::CallbackSettings()
{
    static struct http_parser_settings __parser_setting =
        {
            on_message_begin : CallbackOnMessageBegin, /* on_message_begin */
            on_url : nullptr, /* on_url */
            on_status : nullptr, /* on_status */
            on_header_field : CallbackOnHeaderField, /* on_header_field */
            on_header_value : CallbackOnHeaderValue, /* on_header_value */
            on_headers_complete : CallbackOnHeaderComplete, /* on_headers_complete */
            on_body : CallbackOnBody, /* on_body */
            on_message_complete : CallbackOnMessageComplete, /* on_message_complete */
            on_chunk_header : CallbackOnChunkHeader, /* on_chunk_header */
            on_chunk_complete : CallbackOnChunkComplete
        };

    return &__parser_setting;
}

int Http1ResponseParserCallbacks::CallbackOnMessageBegin(http_parser * parser)
{
    auto * resp_ptr = __response_this_ptr(parser);
    resp_ptr->__set_section_state(resp_ptr->kSectionMessage, resp_ptr->kStateReading);
    return 0;
}

int Http1ResponseParserCallbacks::CallbackOnHeaderField(http_parser * parser, const char * at, size_t length)
{
    auto * resp_ptr = __response_this_ptr(parser);
    resp_ptr->__set_section_state(resp_ptr->kSectionHeader, resp_ptr->kStateReading);
    resp_ptr->headers_.ConstructByHttpParserField(at, length);
    return 0;
}

int Http1ResponseParserCallbacks::CallbackOnHeaderValue(http_parser * parser, const char * at, size_t length)
{
    __response_this_ptr(parser)->headers_.ConstructByHttpParserValue(at, length);
    return 0;
}

int Http1ResponseParserCallbacks::CallbackOnHeaderComplete(http_parser * parser)
{
    auto * resp_ptr = __response_this_ptr(parser);
    resp_ptr->__set_section_state(resp_ptr->kSectionHeader, resp_ptr->kStateComplete);

    /* Http Version and Status Code */
    resp_ptr->resp_version_major_ = parser->http_major;
    resp_ptr->resp_version_minor_ = parser->http_minor;
    resp_ptr->resp_code_ = parser->status_code;

    /* Flush Last HTTP Resp Header-Value */
    resp_ptr->headers_.ConstructByHttpParserComplete();

    /* Lookup Content-Type and Content-Length */
    resp_ptr->headers_.ForEachValueOfHeader("Content-Type", [resp_ptr](const std::string & f, const std::string &v)
    {
        resp_ptr->content_type = v;
        return false;
    });

    resp_ptr->headers_.ForEachValueOfHeader("Content-Length", [resp_ptr](const std::string & f, const std::string &v)
    {
        resp_ptr->content_length = v;
        return false;
    });

    resp_ptr->headers_.ForEachValueOfHeader("Content-Encoding", [resp_ptr](const std::string & f, const std::string &v)
    {
        resp_ptr->content_encoding = v;
        return false;
    });

    return 0;
}

int Http1ResponseParserCallbacks::CallbackOnBody(http_parser * parser, const char * at, size_t length)
{
    auto * resp_ptr = __response_this_ptr(parser);
    resp_ptr->__set_section_state(resp_ptr->kSectionBody, resp_ptr->kStateReading);

    auto content_length = parser->flags & F_CONTENTLENGTH ? parser->content_length : 0;

    /* 第一次回调Body，分配内存，预留内存空间  */
    if (resp_ptr->__is_body_content_empty())
    {
        resp_ptr->__new_body_content(content_length);
    }

    auto & __last_body_content = resp_ptr->__last_body_content().ptr;
    __last_body_content->insert(__last_body_content->end(), at, at + length);

    return 0;
}

int Http1ResponseParserCallbacks::CallbackOnMessageComplete(http_parser * parser)
{
    auto * resp_ptr = __response_this_ptr(parser);
    resp_ptr->__set_section_state(resp_ptr->kSectionBody, resp_ptr->kStateComplete);
    resp_ptr->__set_section_state(resp_ptr->kSectionMessage, resp_ptr->kStateComplete);
    return 0;
}

int Http1ResponseParserCallbacks::CallbackOnChunkHeader(http_parser * parser)
{
    auto * resp_ptr = __response_this_ptr(parser);
    resp_ptr->__set_section_state(resp_ptr->kSectionBody, resp_ptr->kStateReading);

    /* On chunk begin, setup chunk tag */
    resp_ptr->body_is_chunk = true;
    resp_ptr->body_chunk_id++;

    /* 增加新的Body分节 */
    resp_ptr->__new_body_content(parser->content_length);
    return 0;
}

int Http1ResponseParserCallbacks::CallbackOnChunkComplete(http_parser * parser)
{
    /* Body状态设置为Complete，允许上层处理 */
    auto * resp_ptr = __response_this_ptr(parser);
    resp_ptr->__set_section_state(resp_ptr->kSectionBody, resp_ptr->kStateComplete);

    /* 最后一个Body分节的状态置为完成，表示该Body分节数据完成 */
    auto & __last_body_content = resp_ptr->__last_body_content();

    /* 断言，该分节不应该是完整的，否则出现了多次调用 */
    assert(!__last_body_content.is_complete);
    __last_body_content.is_complete = true;

    /* 设置Pause模式，暂停解析，以便调用外部业务处理函数 */
    http_parser_pause(parser, 1);
    return 0;
}

Http1Response::Http1Response(short major_version, short minor_version) : resp_version_major_(major_version),
                                                                         resp_version_minor_(minor_version)
{
    http_parser_init(parser_.get(), HTTP_RESPONSE);
}

ssize_t Http1Response::ConstructFromMemory(const char * buf, size_t buflen)
{
    parser_->data = this;

    /* 31 == HTTP_PAUSED, TODO: 替换这一硬编码 */
    if (parser_->http_errno == 31) http_parser_pause(parser_.get(), 0);
    if (bypass_) parser_->flags |= F_SKIPBODY;

    size_t sz_parsed = http_parser_execute(parser_.get(),
        Http1ResponseParserCallbacks::CallbackSettings(), buf, buflen);

    /* 暂停状态，为了调用外部回调函数 */
    if(sz_parsed && parser_->http_errno == 31)
        return sz_parsed;

    /* 解析错误 */
    if (sz_parsed && parser_->http_errno > 0)
    {
        throw std::runtime_error(string_format("Failed at http parsing: errcode=%u, %s, %s",
            parser_->http_errno, http_errno_name(static_cast<http_errno>(parser_->http_errno)),
            http_errno_description(static_cast<http_errno>(parser_->http_errno))));
    }

    return sz_parsed;
}

ssize_t Http1Response::ConstructFromEvBuf(struct evbuffer * evbuf_ptr)
{
    /* 展平输入的Buffer为线性空间 */
    auto * __data_ptr = (const char *) (evbuffer_pullup(evbuf_ptr, -1));
    size_t __data_len = evbuffer_get_length(evbuf_ptr);

    /* 解析输入的Buffer */
    ssize_t forward_len = ConstructFromMemory(__data_ptr, __data_len);
    if (forward_len < 0) return forward_len;

    /* 已解析的部分，从原Buffer中抽离出来 */
    struct evbuffer * reserved_buffer = evbuffer_new();
    evbuffer_remove_buffer(evbuf_ptr, reserved_buffer, static_cast<size_t>(forward_len));

    /* 保存在上下文中 */
    if (evbuf_content_raw_ == nullptr)
    {
        evbuf_content_raw_ = evbuffer_unique_ptr_t(reserved_buffer);
    }
    else
    {
        evbuffer_add_buffer(evbuf_content_raw_.get(), reserved_buffer);
        evbuffer_free(reserved_buffer);
    }

    return forward_len;
}

void Http1Response::Construct()
{
    /* 新申请EvBuffer */
    evbuffer_unique_ptr_t evbuf_construct = evbuffer_unique_ptr_t(evbuffer_new());
    struct evbuffer * evbuf_ptr = evbuf_construct.get();

    /* 应答第一行 */
    evbuffer_add_printf(evbuf_ptr, "HTTP/%d.%d %d %s\r\n",
        resp_version_major_, resp_version_minor_, resp_code_, __resp_code_to_str(resp_code_));

    /* 应答头部 */
    headers_.ForEachHeader([this, evbuf_ptr](const std::string & str_field, const std::string & str_value)
    {
        evbuffer_add_printf(evbuf_ptr, "%s: %s\r\n", str_field.c_str(), str_value.c_str());
        return true;
    });

    /* 结尾行 */
    evbuffer_add_printf(evbuf_ptr, "\r\n");

    for(auto & body_segment : body_contents_)
    {
        if (body_segment.ptr == nullptr || !body_segment.is_complete) continue;
        auto * body_segment_ptr = body_segment.ptr.get();

        if (body_is_chunk)
        {
            evbuffer_add_printf(evbuf_ptr, "%lx\r\n", body_segment_ptr->size());
            evbuffer_add(evbuf_ptr, body_segment_ptr->data(), body_segment.ptr->size());
            evbuffer_add_printf(evbuf_ptr, "\r\n");
        }
        else
        {
            evbuffer_add(evbuf_ptr, body_segment.ptr->data(), body_segment.ptr->size());
        }
    }

    evbuf_content_raw_ = std::move(evbuf_construct);
    section_state[kSectionHeader] = kStateComplete;
    section_state[kSectionBody] = kStateComplete;
    section_state[kSectionMessage] = kStateComplete;

    return;
}

evbuffer_unique_ptr_t Http1Response::StolenEvBuf()
{
    if (SectionState(kSectionHeader) == kStateComplete)
    {
        __set_section_state(kSectionHeader, kStateStolen);
    }

    if (SectionState(kSectionBody) == kStateComplete)
    {
        __set_section_state(kSectionBody, kStateStolen);
        __drop_all_body_content();
    }

    if (SectionState(kSectionMessage) == kStateComplete)
    {
        __set_section_state(kSectionMessage, kStateStolen);
        __drop_all_body_content();
    }

    return std::move(evbuf_content_raw_);
}

std::string Http1Response::DumpToString()
{
    return {std::to_string(resp_code_) + " " + content_type + " " + content_length + " " + content_encoding};
}

int Http1Connection::on_connection_close(pxy_conn_ctx_t * conn_ctx, struct bufferevent * bev)
{
    return 0;
}

int Http1Connection::on_connection_read_request(pxy_conn_ctx_t * conn_ctx, pxy_conn_desc_t * conn_this,
    pxy_conn_desc_t * conn_other)
{
    /* Get the last session */
    auto * http_session = last_uncomplete_session(kDirectionRequest);
    assert(http_session != nullptr);

    auto & request = dynamic_cast<Http1Request &>(http_session->request());
    auto * downstream_evbuf = bufferevent_get_input(conn_this->bev);
    auto * upstream_evbuf = bufferevent_get_output(conn_other->bev);

    ssize_t forward_len = 0;
    try
    {
        forward_len = request.ConstructFromEvBuf(downstream_evbuf);
    }
    catch (std::runtime_error & err)
    {
        auto str_src_addr = sockaddr_to_string(SockAddrSource());
        auto str_dst_addr = sockaddr_to_string(SockAddrDest());

        LOG(ERROR) << str_src_addr << str_dst_addr << "PASSTHROUGH" << err.what();
        conn_ctx->passthrough = 1;
        return 0;
    }

    if (forward_len < 0)
    {
        conn_ctx->passthrough = 1;
        return 0;
    }

    if (request.Complete(HttpRequest::kSectionHeader))
    {
        http_session->CallRequestHeaderCallback();
    }

    if (request.Complete(HttpRequest::kSectionBody) && request.Body() != nullptr)
    {
        http_session->CallRequestBodyCallback();
    }

    if (http_session->NeedToDrop())
    {
        drop_last_session();
        return 0;
    }

    if (request.ReadOnly() || request.Complete(HttpRequest::kSecionMessage))
    {
        auto __ev_buffer_ptr = request.StolenEvbuf();
        evbuffer_add_buffer(upstream_evbuf, __ev_buffer_ptr.get());
    }

    return 0;
}

int Http1Connection::on_connection_read_response(pxy_conn_ctx_t * conn_ctx, pxy_conn_desc_t * conn_this,
    pxy_conn_desc_t * conn_other)
{
    auto * session = last_uncomplete_session(kDirectionResponse);
    auto * downstream_evbuf = bufferevent_get_input(conn_this->bev);
    auto * upstream_evbuf = bufferevent_get_output(conn_other->bev);

    if (session == nullptr)
    {
        LOG(WARNING) << "Found standlone HTTP response, Ignore it.";
        LOG(WARNING) << hexdump("upstream recieve buffer", downstream_evbuf);
        return 0;
    }

    auto & response = dynamic_cast<Http1Response &>(session->response());

    /* 检查Bypass和Drop标记，该标记可能由请求侧设置
     * 设置任何一个，即不调用上层业务函数，直接解析到消息结束 */
    if (session->NeedToBypass() || session->NeedToDrop())
    {
        response.Bypass(true);
    }

    ssize_t forward_len = 0;
    try
    {
        forward_len = response.ConstructFromEvBuf(downstream_evbuf);
    }
    catch (std::runtime_error & err)
    {
        auto str_src_addr = sockaddr_to_string(SockAddrSource());
        auto str_dst_addr = sockaddr_to_string(SockAddrDest());

        LOG(ERROR) << str_src_addr << str_dst_addr << "PASSTHROUGH" << err.what();
        conn_ctx->passthrough = 1;
        return 0;
    }

    if (forward_len <= 0) return 0;

    if (response.SectionState(response.kSectionHeader) == response.kStateComplete)
    {
        session->CallResponseHeaderCallback();
    }

    if (session->NeedToBypass() || session->NeedToDrop())
    {
        response.Bypass(true);
        goto __forward;
    }

    if (response.SectionState(response.kSectionBody) == response.kStateComplete)
    {
        session->CallResponseBodyCallback();
    }

    if (session->NeedToBypass() || session->NeedToDrop())
    {
        response.Bypass(true);
        goto __forward;
    }

    /* 如果有任何一部分为Reading，说明该部分数据还为到来，暂时不转发 */
    if (response.SectionState(response.kSectionHeader) == response.kStateReading ||
        response.SectionState(response.kSectionBody) == response.kStateReading)
    {
        return 0;
    }

__forward:
    auto __evbuf_ptr = response.StolenEvBuf();
    if (!session->NeedToDrop())
    {
        evbuffer_add_buffer(upstream_evbuf, __evbuf_ptr.get());
    }

    if (response.SectionState(response.kSectionMessage) == response.kStateComplete ||
        response.SectionState(response.kSectionMessage) == response.kStateStolen)
    {
        drop_first_session();
    }

    return 0;
}

HttpSession * Http1Connection::create_new_session()
{
    /* Create new session, set a new request */
    auto __http_session = std::unique_ptr<HttpSession>(new HttpSession(*this));
    __http_session->request(HttpRequestFactory(1, 0));
    __http_session->response(HttpResponseFactory(1, 0));

    /* Callback on new session */
    if (session_new_cb_) session_new_cb_(*__http_session);

    /* Add to the last record */
    http_sessions_.push_back(std::move(__http_session));
    return http_sessions_.back().get();
}

HttpSession * Http1Connection::last_uncomplete_session(enum direction dir)
{
    if (dir == kDirectionRequest)
    {
        if (http_sessions_.empty())
        {
            return create_new_session();
        }

        /* 看一下请求侧的状态，如果已经处理完了，说明使用了HttpPipeLine，新建一个Session */
        auto * __session = http_sessions_.back().get();
        if (__session->request().Complete(HttpRequest::kSecionMessage))
            return create_new_session();

        /* 否则，返回最后一个没有完全处理结束的Session */
        return __session;
    }
    else if (dir == kDirectionResponse)
    {
        if (http_sessions_.empty())
        {
            LOG(DEBUG) << "session queue is empty, no corresponding session for response.";
            return nullptr;
        }

        auto * __session = http_sessions_.front().get();
        return __session;
    }

    assert(0);
    return nullptr;
}

void Http1Connection::drop_last_session()
{
    assert(http_sessions_.cbegin() != http_sessions_.cend());
    http_sessions_.pop_back();
}

void Http1Connection::drop_first_session()
{
    assert(http_sessions_.cbegin() != http_sessions_.cend());
    http_sessions_.pop_front();
}

const sockaddr * Http1Connection::SockAddrSource() const
{
    return (const sockaddr *) &sockaddr_source_;
}

const sockaddr * Http1Connection::SockAddrDest() const
{
    return (const sockaddr *) &sockaddr_dest_;
}

void Http1Connection::Write(std::unique_ptr<HttpSession> http_session)
{
    auto & request = dynamic_cast<Http1Request &>(http_session->request());
    auto & response = dynamic_cast<Http1Response &>(http_session->response());

    if (request.Complete(HttpRequest::kSecionMessage))
    {
        auto stolen_buffer = request.StolenEvbuf();
        bufferevent_write_buffer(bev_upstream_, stolen_buffer.get());
    }

    if (response.SectionState(response.kSectionMessage) == response.kStateComplete)
    {
        auto stolen_buffer = response.StolenEvBuf();
        bufferevent_write_buffer(bev_downstream_, stolen_buffer.get());
    }

    return;
}

std::unique_ptr<HttpRequest> HttpRequestFactory(short major_version, short minor_version)
{
    if (major_version == 1 && minor_version == 0)
        return std::move(std::make_unique<Http1Request>(major_version, minor_version));
    if (major_version == 1 && minor_version == 1)
        return std::move(std::make_unique<Http1Request>(major_version, minor_version));

    throw std::invalid_argument(string_format("Invalid HTTP Version: %d, %d",
        major_version, minor_version));
}

std::unique_ptr<HttpResponse> HttpResponseFactory(short major_version, short minor_version)
{
    if (major_version == 1 && minor_version == 0)
        return std::move(std::make_unique<Http1Response>(major_version, minor_version));
    if (major_version == 1 && minor_version == 1)
        return std::move(std::make_unique<Http1Response>(major_version, minor_version));

    throw std::invalid_argument(string_format("Invalid HTTP Version: %d, %d",
        major_version, minor_version));
}