update http client (#317)

2023-06-09 17:15:22 +08:00 · 2023-06-09 17:15:22 +08:00 · 3edd2943b2
parent 18f4d5626a
commit 3edd2943b2
8 changed files with 175 additions and 56 deletions
--- a/.github/workflows/ubuntu_clang.yml
+++ b/.github/workflows/ubuntu_clang.yml
@ -129,7 +129,7 @@ jobs:
                -DENABLE_FILE_IO_URING=${{matrix.io_uring}} \
                -DUSE_CCACHE=${{env.ccache}}
      - name: Build
-        run: cmake --build ${{github.workspace}}/build --config ${{matrix.mode}}
+        run: cmake --build ${{github.workspace}}/build --config ${{matrix.mode}} --target ${{matrix.case_name}}

      - name: Test
        working-directory: ${{github.workspace}}/build
--- a/.github/workflows/ubuntu_gcc.yml
+++ b/.github/workflows/ubuntu_gcc.yml
@ -125,7 +125,7 @@ jobs:
                -DUSE_CCACHE=${{env.ccache}}

      - name: Build
-        run: cmake --build ${{github.workspace}}/build --config ${{matrix.mode}}
+        run: cmake --build ${{github.workspace}}/build --config ${{matrix.mode}} --target ${{matrix.case_name}}

      - name: Test
        working-directory: ${{github.workspace}}/build
--- a/include/coro_io/coro_file.hpp
+++ b/include/coro_io/coro_file.hpp
@ -15,6 +15,7 @@
 */
 #pragma once
 #include <asio/io_context.hpp>
+#include <filesystem>
 #include <fstream>

 #include "async_simple/Promise.h"
@ -51,8 +52,12 @@ enum class open_mode { read, write };
 class coro_file {
 public:
 #if defined(ENABLE_FILE_IO_URING)
-  coro_file(asio::io_context::executor_type executor,
-            const std::string& filepath, open_mode flags = open_mode::read) {
+  coro_file(coro_io::ExecutorWrapper<>* executor, std::string_view filepath,
+            open_mode flags = open_mode::read)
+      : coro_file(executor->get_asio_executor(), filepath, flags) {}
+
+  coro_file(asio::io_context::executor_type executor, std::string_view filepath,
+            open_mode flags = open_mode::read) {
    try {
      stream_file_ = std::make_unique<asio::stream_file>(executor);
    } catch (std::exception& ex) {
@ -61,20 +66,25 @@ class coro_file {
    }

    std::error_code ec;
-    stream_file_->open(filepath, default_flags(), ec);
+    stream_file_->open(filepath.data(), default_flags(), ec);
    if (ec) {
      std::cout << ec.message() << "\n";
    }
  }
 #else

-  coro_file(asio::io_context::executor_type executor,
-            const std::string& filepath, open_mode flags = open_mode::read)
+  coro_file(coro_io::ExecutorWrapper<>* executor, std::string_view filepath,
+            open_mode flags = open_mode::read)
+      : coro_file(executor->get_asio_executor(), filepath, flags) {}
+
+  coro_file(asio::io_context::executor_type executor, std::string_view filepath,
+            open_mode flags = open_mode::read)
      : executor_wrapper_(executor) {
    std::ios::openmode open_flags = flags == open_mode::read
                                        ? std::ios::binary | std::ios::in
                                        : std::ios::out | std::ios::app;
-    stream_file_ = std::make_unique<std::fstream>(filepath, open_flags);
+    stream_file_ = std::make_unique<std::fstream>(
+        std::filesystem::path(filepath), open_flags);
    if (!stream_file_->is_open()) {
      std::cout << "open file " << filepath << " failed "
                << "\n";
@ -95,6 +105,12 @@ class coro_file {

  void close() { stream_file_.reset(); }

+  static size_t file_size(std::string_view filepath) {
+    std::error_code ec;
+    size_t size = std::filesystem::file_size(filepath, ec);
+    return size;
+  }
+
 #if defined(ENABLE_FILE_IO_URING)
  async_simple::coro::Lazy<std::pair<std::error_code, size_t>> async_read(
      char* data, size_t size) {
@ -226,8 +242,9 @@ class coro_file {
  std::atomic<size_t> seek_offset_ = 0;
 #else
  std::unique_ptr<std::fstream> stream_file_;
-  coro_io::ExecutorWrapper<asio::io_context::executor_type> executor_wrapper_;
+  coro_io::ExecutorWrapper<> executor_wrapper_;
 #endif
+
  std::atomic<bool> eof_ = false;
 };
 }  // namespace coro_io
--- a/src/coro_http/examples/example.cpp
+++ b/src/coro_http/examples/example.cpp
@ -119,11 +119,11 @@ async_simple::coro::Lazy<void> upload_files(
  client.add_str_part("key", "value");
  client.add_file_part("test", "test.jpg");
  std::string uri = "http://yoururl.com";
-  auto result = co_await client.async_upload(uri);
+  auto result = co_await client.async_upload_multipart(uri);
  std::cout << result.net_err << "\n";
  std::cout << result.status << "\n";

-  result = co_await client.async_upload(uri, "test", "test.jpg");
+  result = co_await client.async_upload_multipart(uri, "test", "test.jpg");
  std::cout << result.status << "\n";
 }

--- a/thirdparty/cinatra/cinatra/coro_http_client.hpp
+++ b/thirdparty/cinatra/cinatra/coro_http_client.hpp
@ -18,6 +18,7 @@
 #include "async_simple/Future.h"
 #include "async_simple/coro/FutureAwaiter.h"
 #include "async_simple/coro/Lazy.h"
+#include "coro_io/coro_file.hpp"
 #include "coro_io/coro_io.hpp"
 #include "http_parser.hpp"
 #include "response_cv.hpp"
@ -110,6 +111,9 @@ class coro_http_client {
        executor_wrapper_(executor),
        timer_(&executor_wrapper_) {}

+  coro_http_client(coro_io::ExecutorWrapper<> *executor)
+      : coro_http_client(executor->get_asio_executor()) {}
+
  bool init_config(const config &conf) {
    if (conf.conn_timeout_duration.has_value()) {
      set_conn_timeout(*conf.conn_timeout_duration);
@ -349,7 +353,8 @@ class coro_http_client {
                         cinatra::req_context<>{});
  }

-  async_simple::coro::Lazy<resp_data> async_connect(std::string uri) {
+  // CONNECT example.com HTTP/1.1
+  async_simple::coro::Lazy<resp_data> async_http_connect(std::string uri) {
    return async_request(std::move(uri), cinatra::http_method::CONNECT,
                         cinatra::req_context<>{});
  }
@ -533,7 +538,7 @@ class coro_http_client {
    co_return std::error_code{};
  }

-  async_simple::coro::Lazy<resp_data> async_upload(std::string uri) {
+  async_simple::coro::Lazy<resp_data> async_upload_multipart(std::string uri) {
    std::shared_ptr<int> guard(nullptr, [this](auto) {
      req_headers_.clear();
      form_data_.clear();
@ -614,16 +619,15 @@ class coro_http_client {
    co_return data;
  }

-  async_simple::coro::Lazy<resp_data> async_upload(std::string uri,
-                                                   std::string name,
-                                                   std::string filename) {
+  async_simple::coro::Lazy<resp_data> async_upload_multipart(
+      std::string uri, std::string name, std::string filename) {
    if (!add_file_part(std::move(name), std::move(filename))) {
 #ifndef NDEBUG
      std::cout << "open file failed or duplicate test names\n";
 #endif
      co_return resp_data{{}, 404};
    }
-    co_return co_await async_upload(std::move(uri));
+    co_return co_await async_upload_multipart(std::move(uri));
  }

  async_simple::coro::Lazy<resp_data> async_download(std::string uri,
@ -678,6 +682,75 @@ class coro_http_client {
    co_return co_await connect(std::move(uri));
  }

+  async_simple::coro::Lazy<resp_data> async_upload_chunked(
+      std::string uri, http_method method, std::string filename,
+      std::unordered_map<std::string, std::string> headers = {}) {
+    std::shared_ptr<int> guard(nullptr, [this](auto) {
+      if (!req_headers_.empty()) {
+        req_headers_.clear();
+      }
+    });
+
+    req_context<> ctx{req_content_type::text};
+    resp_data data{};
+    auto [ok, u] = handle_uri(data, uri);
+    if (!ok) {
+      co_return resp_data{{}, 404};
+    }
+
+    if (!std::filesystem::exists(filename)) {
+      co_return resp_data{
+          std::make_error_code(std::errc::no_such_file_or_directory), 404};
+    }
+
+    add_header("Transfer-Encoding", "chunked");
+
+    std::string header_str =
+        build_request_header(u, method, ctx, true, std::move(headers));
+    std::cout << header_str;
+
+    std::error_code ec{};
+    size_t size = 0;
+
+    auto promise = start_timer(req_timeout_duration_, "connect timer");
+
+    data = co_await connect(u);
+    if (ec = co_await wait_timer(promise); ec) {
+      co_return resp_data{{}, 404};
+    }
+    if (data.net_err) {
+      co_return data;
+    }
+
+    promise = start_timer(req_timeout_duration_, "upload timer");
+    std::tie(ec, size) = co_await async_write(asio::buffer(header_str));
+    if (ec) {
+      co_return resp_data{ec, 404};
+    }
+
+    coro_io::coro_file file(executor_wrapper_, filename);
+    char buf[4096];
+    std::string chunk_size_str;
+    while (!file.eof()) {
+      auto [rd_ec, rd_size] = co_await file.async_read(buf, 4096);
+      auto bufs = cinatra::to_chunked_buffers<asio::const_buffer>(
+          buf, rd_size, chunk_size_str, file.eof());
+      if (std::tie(ec, size) = co_await async_write(bufs); ec) {
+        co_return resp_data{ec, 404};
+      }
+    }
+
+    bool is_keep_alive = true;
+    data = co_await handle_read(ec, size, is_keep_alive, std::move(ctx),
+                                http_method::POST);
+    if (auto errc = co_await wait_timer(promise); errc) {
+      ec = errc;
+    }
+
+    handle_result(data, ec, is_keep_alive);
+    co_return data;
+  }
+
  async_simple::coro::Lazy<resp_data> async_request(
      std::string uri, http_method method, auto ctx,
      std::unordered_map<std::string, std::string> headers = {}) {
@ -862,6 +935,7 @@ class coro_http_client {

  std::string build_request_header(
      const uri_t &u, http_method method, const auto &ctx,
+      bool is_chunked = false,
      std::unordered_map<std::string, std::string> headers = {}) {
    std::string req_str(method_name(method));

@ -926,8 +1000,13 @@ class coro_http_client {
    }
    else {
      if ((method == http_method::POST || method == http_method::PUT) &&
-          ctx.content_type != req_content_type::multipart)
+          ctx.content_type != req_content_type::multipart) {
        should_add = true;
+      }
+    }
+
+    if (is_chunked) {
+      should_add = false;
    }

    if (should_add) {
@ -946,7 +1025,7 @@ class coro_http_client {
      const uri_t &u, http_method method, const auto &ctx,
      std::unordered_map<std::string, std::string> headers) {
    std::string req_str =
-        build_request_header(u, method, ctx, std::move(headers));
+        build_request_header(u, method, ctx, false, std::move(headers));

 #ifdef CORO_HTTP_PRINT_REQ_HEAD
    std::cout << req_str << "\n";
--- a/thirdparty/cinatra/cinatra/define.h
+++ b/thirdparty/cinatra/cinatra/define.h
@ -18,6 +18,7 @@ enum class content_type {
 enum class req_content_type {
  html,
  json,
+  text,
  string,
  multipart,
  ranges,
--- a/thirdparty/cinatra/cinatra/response_cv.hpp
+++ b/thirdparty/cinatra/cinatra/response_cv.hpp
@ -186,8 +186,8 @@ inline constexpr std::string_view rep_server = "Server: cinatra\r\n";
 inline const char name_value_separator[] = {':', ' '};
 // inline std::string_view crlf = "\r\n";

-inline const char crlf[] = {'\r', '\n'};
-inline const char last_chunk[] = {'0', '\r', '\n'};
+constexpr std::string_view crlf = "\r\n";
+constexpr std::string_view last_chunk = "0\r\n";
 inline const std::string http_chunk_header =
    "HTTP/1.1 200 OK\r\n"
    "Transfer-Encoding: chunked\r\n";
@ -200,7 +200,7 @@ inline const std::string http_range_chunk_header =
 /*"Content-Type: video/mp4\r\n"
 "\r\n";*/

-inline constexpr auto to_content_type_str(req_content_type type) {
+inline constexpr std::string_view to_content_type_str(req_content_type type) {
  switch (type) {
    case req_content_type::html:
      return rep_html;
@ -211,7 +211,7 @@ inline constexpr auto to_content_type_str(req_content_type type) {
    case req_content_type::multipart:
      return rep_multipart;
    default:
-      return ""sv;
+      return "";
  }
 }

@ -254,60 +254,54 @@ struct explode<0, digits...> : to_chars<digits...> {};
 template <unsigned num>
 struct num_to_string : detail::explode<num / 10, num % 10> {};

-inline asio::const_buffer to_buffer(status_type status) {
+template <typename T>
+inline decltype(auto) to_buffer(status_type status) {
  switch (status) {
    case status_type::switching_protocols:
-      return asio::buffer(switching_protocols.data(),
-                          switching_protocols.length());
+      return T(switching_protocols.data(), switching_protocols.length());
    case status_type::ok:
-      return asio::buffer(rep_ok.data(), rep_ok.length());
+      return T(rep_ok.data(), rep_ok.length());
    case status_type::created:
-      return asio::buffer(rep_created.data(), rep_created.length());
+      return T(rep_created.data(), rep_created.length());
    case status_type::accepted:
-      return asio::buffer(rep_accepted.data(), rep_created.length());
+      return T(rep_accepted.data(), rep_created.length());
    case status_type::no_content:
-      return asio::buffer(rep_no_content.data(), rep_no_content.length());
+      return T(rep_no_content.data(), rep_no_content.length());
    case status_type::partial_content:
-      return asio::buffer(rep_partial_content.data(),
-                          rep_partial_content.length());
+      return T(rep_partial_content.data(), rep_partial_content.length());
    case status_type::multiple_choices:
-      return asio::buffer(rep_multiple_choices.data(),
-                          rep_multiple_choices.length());
+      return T(rep_multiple_choices.data(), rep_multiple_choices.length());
    case status_type::moved_permanently:
-      return asio::buffer(rep_moved_permanently.data(),
-                          rep_moved_permanently.length());
+      return T(rep_moved_permanently.data(), rep_moved_permanently.length());
    case status_type::temporary_redirect:
-      return asio::buffer(rep_temporary_redirect.data(),
-                          rep_temporary_redirect.length());
+      return T(rep_temporary_redirect.data(), rep_temporary_redirect.length());
    case status_type::moved_temporarily:
-      return asio::buffer(rep_moved_temporarily.data(),
-                          rep_moved_temporarily.length());
+      return T(rep_moved_temporarily.data(), rep_moved_temporarily.length());
    case status_type::not_modified:
-      return asio::buffer(rep_not_modified.data(), rep_not_modified.length());
+      return T(rep_not_modified.data(), rep_not_modified.length());
    case status_type::bad_request:
-      return asio::buffer(rep_bad_request.data(), rep_bad_request.length());
+      return T(rep_bad_request.data(), rep_bad_request.length());
    case status_type::unauthorized:
-      return asio::buffer(rep_unauthorized.data(), rep_unauthorized.length());
+      return T(rep_unauthorized.data(), rep_unauthorized.length());
    case status_type::forbidden:
-      return asio::buffer(rep_forbidden.data(), rep_forbidden.length());
+      return T(rep_forbidden.data(), rep_forbidden.length());
    case status_type::not_found:
-      return asio::buffer(rep_not_found.data(), rep_not_found.length());
+      return T(rep_not_found.data(), rep_not_found.length());
    case status_type::conflict:
-      return asio::buffer(rep_conflict.data(), rep_conflict.length());
+      return T(rep_conflict.data(), rep_conflict.length());
    case status_type::internal_server_error:
-      return asio::buffer(rep_internal_server_error.data(),
-                          rep_internal_server_error.length());
+      return T(rep_internal_server_error.data(),
+               rep_internal_server_error.length());
    case status_type::not_implemented:
-      return asio::buffer(rep_not_implemented.data(),
-                          rep_not_implemented.length());
+      return T(rep_not_implemented.data(), rep_not_implemented.length());
    case status_type::bad_gateway:
-      return asio::buffer(rep_bad_gateway.data(), rep_bad_gateway.length());
+      return T(rep_bad_gateway.data(), rep_bad_gateway.length());
    case status_type::service_unavailable:
-      return asio::buffer(rep_service_unavailable.data(),
-                          rep_service_unavailable.length());
+      return T(rep_service_unavailable.data(),
+               rep_service_unavailable.length());
    default:
-      return asio::buffer(rep_internal_server_error.data(),
-                          rep_internal_server_error.length());
+      return T(rep_internal_server_error.data(),
+               rep_internal_server_error.length());
  }
 }

--- a/thirdparty/cinatra/cinatra/utils.hpp
+++ b/thirdparty/cinatra/cinatra/utils.hpp
@ -18,9 +18,9 @@
 #include <string>
 #include <string_view>
 #include <type_traits>
-#include <vector>

 #include "define.h"
+#include "response_cv.hpp"
 #include "sha1.hpp"

 namespace cinatra {
@ -267,6 +267,9 @@ inline std::string get_content_type_str(req_content_type type) {
    case req_content_type::json:
      str = "application/json; charset=UTF-8";
      break;
+    case req_content_type::text:
+      str = "text/plain";
+      break;
    case req_content_type::string:
      str = "text/html; charset=UTF-8";
      break;
@ -514,6 +517,31 @@ inline int64_t hex_to_int(std::string_view s) {
  return n;
 }

+template <typename T>
+inline std::vector<T> to_chunked_buffers(const char *chunk_data, size_t length,
+                                         std::string &chunk_size, bool eof) {
+  std::vector<T> buffers;
+
+  if (length > 0) {
+    // convert bytes transferred count to a hex string.
+    chunk_size = to_hex_string(length);
+
+    // Construct chunk based on rfc2616 section 3.6.1
+    buffers.push_back(T(chunk_size.data(), chunk_size.size()));
+    buffers.push_back(T(crlf.data(), crlf.size()));
+    buffers.push_back(T(chunk_data, length));
+    buffers.push_back(T(crlf.data(), crlf.size()));
+  }
+
+  // append last-chunk
+  if (eof) {
+    buffers.push_back(T(last_chunk.data(), last_chunk.size()));
+    buffers.push_back(T(crlf.data(), crlf.size()));
+  }
+
+  return buffers;
+}
+
 static const std::string base64_chars =
    "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
    "abcdefghijklmnopqrstuvwxyz"