#88 http support chunked

README-CN.md

@@ -22,11 +22,12 @@
 ## ✨ 特征
 
 - 跨平台（Linux, Windows, MacOS, Solaris）
-- 高性能事件循环（网络IO事件、定时器事件、空闲事件）
+- 高性能事件循环（网络IO事件、定时器事件、空闲事件、自定义事件）
 - TCP/UDP服务端/客户端/代理
-- SSL/TLS加密通信（WITH_OPENSSL or WITH_MBEDTLS）
-- HTTP服务端/客户端（https http1/x http2 grpc）
-- HTTP文件服务、目录服务、API服务（支持RESTful）
+- SSL/TLS加密通信（可选WITH_OPENSSL or WITH_MBEDTLS）
+- HTTP服务端/客户端（支持https http1/x http2 grpc）
+- HTTP支持静态文件服务、目录服务、同步/异步API处理函数
+- HTTP支持RESTful风格、URI路由、keep-alive长连接、chunked分块等特性
 - WebSocket服务端/客户端
 
 ## ⌛️ 构建

README.md

@@ -24,11 +24,12 @@ but simpler api and richer protocols.
 ## ✨ Features
 
 - Cross-platform (Linux, Windows, MacOS, Solaris)
-- EventLoop (IO, timer, idle)
+- EventLoop (IO, timer, idle, custom)
 - TCP/UDP client/server/proxy
-- SSL/TLS support: WITH_OPENSSL or WITH_MBEDTLS
-- HTTP client/server (include https http1/x http2 grpc)
-- HTTP file service, indexof service, api service (support RESTful)
+- SSL/TLS support: (via WITH_OPENSSL or WITH_MBEDTLS)
+- HTTP client/server (support https http1/x http2 grpc)
+- HTTP static file service, indexof service, sync/async API handler
+- HTTP supports RESTful, URI router, keep-alive, chunked, etc.
 - WebSocket client/server
 
 ## ⌛️ Build

examples/httpd/handler.h

@@ -1,7 +1,8 @@
 #ifndef HV_HTTPD_HANDLER_H
 #define HV_HTTPD_HANDLER_H
 
-#include <future> // import std::async
+#include <thread>   // import std::thread
+#include <chrono>   // import std::chrono
 
 #include "hbase.h"
 #include "htime.h"
@@ -58,7 +59,7 @@ public:
     }
 
     static int largeFileHandler(const HttpContextPtr& ctx) {
-        std::async([ctx](){
+        std::thread([ctx](){
             ctx->writer->Begin();
             std::string filepath = ctx->service->document_root + ctx->request->Path();
             HFile file;
@@ -80,12 +81,16 @@ public:
             size_t filesize = file.size();
             ctx->writer->WriteHeader("Content-Type", http_content_type_str(content_type));
             ctx->writer->WriteHeader("Content-Length", filesize);
+            // ctx->writer->WriteHeader("Transfer-Encoding", "chunked");
             ctx->writer->EndHeaders();
 
             char* buf = NULL;
             int len = 4096; // 4K
             SAFE_ALLOC(buf, len);
             size_t total_readbytes = 0;
+            int last_progress = 0;
+            auto start_time = std::chrono::steady_clock::now();
+            auto end_time = start_time;
             while (total_readbytes < filesize) {
                 size_t readbytes = file.read(buf, len);
                 if (readbytes <= 0) {
@@ -96,10 +101,20 @@ public:
                     break;
                 }
                 total_readbytes += readbytes;
+                int cur_progress = total_readbytes * 100 / filesize;
+                if (cur_progress > last_progress) {
+                    // printf("<< %s progress: %ld/%ld = %d%%\n",
+                    //     ctx->request->path.c_str(), (long)total_readbytes, (long)filesize, (int)cur_progress);
+                    last_progress = cur_progress;
+                }
+                end_time += std::chrono::milliseconds(len / 1024); // 1KB/ms = 1MB/s = 8Mbps
+                std::this_thread::sleep_until(end_time);
             }
             ctx->writer->End();
             SAFE_FREE(buf);
-        });
+            // auto elapsed_time = std::chrono::duration_cast<std::chrono::seconds>(end_time - start_time);
+            // printf("<< %s taked %ds\n", ctx->request->path.c_str(), (int)elapsed_time.count());
+        }).detach();
         return 0;
     }
 

http/Http1Parser.cpp

@@ -10,6 +10,8 @@ static int on_body(http_parser* parser, const char *at, size_t length);
 static int on_message_begin(http_parser* parser);
 static int on_headers_complete(http_parser* parser);
 static int on_message_complete(http_parser* parser);
+static int on_chunk_header(http_parser* parser);
+static int on_chunk_complete(http_parser* parser);
 
 http_parser_settings* Http1Parser::cbs = NULL;
 
@@ -25,6 +27,8 @@ Http1Parser::Http1Parser(http_session_type type) {
         cbs->on_headers_complete = on_headers_complete;
         cbs->on_body             = on_body;
         cbs->on_message_complete = on_message_complete;
+        cbs->on_chunk_header     = on_chunk_header;
+        cbs->on_chunk_complete   = on_chunk_complete;
     }
     http_parser_init(&parser, HTTP_BOTH);
     parser.data = this;
@@ -46,7 +50,7 @@ int on_url(http_parser* parser, const char *at, size_t length) {
 }
 
 int on_status(http_parser* parser, const char *at, size_t length) {
-    printd("on_status:%.*s\n", (int)length, at);
+    printd("on_status:%d %.*s\n", (int)parser->status_code, (int)length, at);
     Http1Parser* hp = (Http1Parser*)parser->data;
     hp->state = HP_STATUS;
     return 0;
@@ -62,7 +66,7 @@ int on_header_field(http_parser* parser, const char *at, size_t length) {
 }
 
 int on_header_value(http_parser* parser, const char *at, size_t length) {
-    printd("on_header_value:%.*s""\n", (int)length, at);
+    printd("on_header_value:%.*s\n", (int)length, at);
     Http1Parser* hp = (Http1Parser*)parser->data;
     hp->state = HP_HEADER_VALUE;
     hp->header_value.append(at, length);
@@ -70,7 +74,8 @@ int on_header_value(http_parser* parser, const char *at, size_t length) {
 }
 
 int on_body(http_parser* parser, const char *at, size_t length) {
-    //printd("on_body:%.*s""\n", (int)length, at);
+    printd("on_body:%d\n", (int)length);
+    // printd("on_body:%.*s\n", (int)length, at);
     Http1Parser* hp = (Http1Parser*)parser->data;
     hp->state = HP_BODY;
     hp->parsed->body.append(at, length);
@@ -129,3 +134,26 @@ int on_message_complete(http_parser* parser) {
     hp->state = HP_MESSAGE_COMPLETE;
     return 0;
 }
+
+int on_chunk_header(http_parser* parser) {
+    printd("on_chunk_header:%llu\n", parser->content_length);
+    Http1Parser* hp = (Http1Parser*)parser->data;
+    hp->state = HP_CHUNK_HEADER;
+    hp->parsed->body.clear();
+    int chunk_size = parser->content_length;
+    int reserve_size = MIN(chunk_size + 1, MAX_CONTENT_LENGTH);
+    if (reserve_size > hp->parsed->body.capacity()) {
+        hp->parsed->body.reserve(reserve_size);
+    }
+    return 0;
+}
+
+int on_chunk_complete(http_parser* parser) {
+    printd("on_chunk_complete\n");
+    Http1Parser* hp = (Http1Parser*)parser->data;
+    hp->state = HP_CHUNK_COMPLETE;
+    if (hp->parsed->chunked_cb) {
+        hp->parsed->chunked_cb(hp->parsed->body.c_str(), hp->parsed->body.size());
+    }
+    return 0;
+}

http/Http1Parser.h

@@ -12,7 +12,9 @@ enum http_parser_state {
     HP_HEADER_FIELD,
     HP_HEADER_VALUE,
     HP_HEADERS_COMPLETE,
+    HP_CHUNK_HEADER,
     HP_BODY,
+    HP_CHUNK_COMPLETE,
     HP_MESSAGE_COMPLETE
 };
 

http/HttpMessage.cpp

@@ -285,19 +285,18 @@ void HttpMessage::FillContentLength() {
     if (iter != headers.end()) {
         content_length = atoi(iter->second.c_str());
     }
-
-    if (iter == headers.end() || content_length == 0) {
-        if (content_length == 0) {
-            content_length = body.size();
-        }
-        if (content_length == 0) {
-            DumpBody();
-            content_length = body.size();
-        }
-        char sz[64];
-        snprintf(sz, sizeof(sz), "%d", content_length);
-        headers["Content-Length"] = sz;
+    if (content_length == 0) {
+        DumpBody();
+        content_length = body.size();
     }
+    if (iter == headers.end() && content_length != 0 && !IsChunked()) {
+        headers["Content-Length"] = hv::to_string(content_length);
+    }
+}
+
+bool HttpMessage::IsChunked() {
+    auto iter = headers.find("Transfer-Encoding");
+    return iter == headers.end() ? false : stricmp(iter->second.c_str(), "chunked") == 0;
 }
 
 bool HttpMessage::IsKeepAlive() {

http/HttpMessage.h

@@ -71,6 +71,7 @@ struct HV_EXPORT HttpCookie {
 typedef std::map<std::string, std::string, StringCaseLess>  http_headers;
 typedef std::vector<HttpCookie>                             http_cookies;
 typedef std::string                                         http_body;
+typedef std::function<void(const char* data, size_t size)>  http_chuncked_cb;
 
 class HV_EXPORT HttpMessage {
 public:
@@ -82,6 +83,7 @@ public:
     http_headers        headers;
     http_cookies        cookies;
     http_body           body;
+    http_chuncked_cb    chunked_cb;
 
     // structured content
     void*               content;    // DATA_NO_COPY
@@ -177,6 +179,7 @@ public:
         content = NULL;
         content_length = 0;
         content_type = CONTENT_TYPE_NONE;
+        chunked_cb = NULL;
     }
 
     virtual void Reset() {
@@ -195,14 +198,12 @@ public:
     // body.size -> content_length <-> headers Content-Length
     void FillContentLength();
 
+    bool IsChunked();
     bool IsKeepAlive();
 
     std::string GetHeader(const char* key, const std::string& defvalue = "") {
         auto iter = headers.find(key);
-        if (iter != headers.end()) {
-            return iter->second;
-        }
-        return defvalue;
+        return iter == headers.end() ? defvalue : iter->second;
     }
 
     // headers -> string
@@ -324,10 +325,7 @@ public:
 
     std::string Host() {
         auto iter = headers.find("Host");
-        if (iter != headers.end()) {
-            host = iter->second;
-        }
-        return host;
+        return iter == headers.end() ? host : iter->second;
     }
 
     std::string Path() {
@@ -339,10 +337,7 @@ public:
 
     std::string GetParam(const char* key, const std::string& defvalue = "") {
         auto iter = query_params.find(key);
-        if (iter != query_params.end()) {
-            return iter->second;
-        }
-        return defvalue;
+        return iter == query_params.end() ? defvalue : iter->second;
     }
 
     // Range: bytes=0-4095

http/server/HttpResponseWriter.h

@@ -13,6 +13,8 @@ public:
         SEND_BEGIN,
         SEND_HEADER,
         SEND_BODY,
+        SEND_CHUNKED,
+        SEND_CHUNKED_END,
         SEND_END,
     } state;
     HttpResponseWriter(hio_t* io, const HttpResponsePtr& resp)
@@ -25,7 +27,8 @@ public:
     // Begin -> End
     // Begin -> WriteResponse -> End
     // Begin -> WriteStatus -> WriteHeader -> WriteBody -> End
-    // Begin -> WriteHeader -> EndHeaders -> WriteBody -> WriteBody -> ... -> End
+    // Begin -> EndHeaders("Content-Length", content_length) -> WriteBody -> WriteBody -> ... -> End
+    // Begin -> EndHeaders("Transfer-Encoding", "chunked") -> WriteChunked -> WriteChunked -> ... -> End
 
     int Begin() {
         state = SEND_BEGIN;
@@ -58,7 +61,44 @@ public:
         return write(headers);
     }
 
+    template<typename T>
+    int EndHeaders(const char* key, T num) {
+        std::string value = hv::to_string(num);
+        return EndHeaders(key, value.c_str());
+    }
+
+    int WriteChunked(const char* buf, int len = -1) {
+        int ret = 0;
+        if (len == -1) len = strlen(buf);
+        if (state == SEND_BEGIN) {
+            EndHeaders("Transfer-Encoding", "chunked");
+        }
+        char chunked_header[64];
+        int chunked_header_len = snprintf(chunked_header, sizeof(chunked_header), "%x\r\n", len);
+        write(chunked_header, chunked_header_len);
+        if (buf && len) {
+            ret = write(buf, len);
+            state = SEND_CHUNKED;
+        } else {
+            state = SEND_CHUNKED_END;
+        }
+        write("\r\n", 2);
+        return ret;
+    }
+
+    int WriteChunked(const std::string& str) {
+        return WriteChunked(str.c_str(), str.size());
+    }
+
+    int EndChunked() {
+        return WriteChunked(NULL, 0);
+    }
+
     int WriteBody(const char* buf, int len = -1) {
+        if (response->IsChunked()) {
+            return WriteChunked(buf, len);
+        }
+
         if (len == -1) len = strlen(buf);
         if (state == SEND_BEGIN) {
             response->body.append(buf, len);
@@ -92,21 +132,31 @@ public:
         }
 
         int ret = 0;
-        if (buf) {
-            ret = WriteBody(buf, len);
-        }
-        bool is_dump_headers = true;
-        bool is_dump_body = true;
-        if (state == SEND_HEADER) {
-            is_dump_headers = false;
-        } else if (state == SEND_BODY) {
-            is_dump_headers = false;
-            is_dump_body = false;
-        }
-        if (is_dump_body) {
-            std::string msg = response->Dump(is_dump_headers, is_dump_body);
-            ret = write(msg);
+        if (state == SEND_CHUNKED) {
+            if (buf) {
+                ret = WriteChunked(buf, len);
+            }
+            if (state == SEND_CHUNKED) {
+                EndChunked();
+            }
+        } else {
+            if (buf) {
+                ret = WriteBody(buf, len);
+            }
+            bool is_dump_headers = true;
+            bool is_dump_body = true;
+            if (state == SEND_HEADER) {
+                is_dump_headers = false;
+            } else if (state == SEND_BODY) {
+                is_dump_headers = false;
+                is_dump_body = false;
+            }
+            if (is_dump_body) {
+                std::string msg = response->Dump(is_dump_headers, is_dump_body);
+                ret = write(msg);
+            }
         }
+
         state = SEND_END;
         if (!response->IsKeepAlive()) {
             close();