[coro_http_client]fix async upload (#870)

Author: qicosmos

include/ylt/metric/metric_manager.hpp

@@ -456,10 +456,11 @@ class dynamic_metric_manager {
  private:
   void clean_label_expired() {
     executor_ = coro_io::create_io_context_pool(1);
+    auto sp = executor_;
     timer_ = std::make_shared<coro_io::period_timer>(executor_->get_executor());
     check_label_expired(timer_)
         .via(executor_->get_executor())
-        .start([](auto&&) {
+        .start([sp](auto&&) {
         });
   }
 

include/ylt/standalone/cinatra/coro_http_client.hpp

@@ -1101,6 +1101,8 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
     req_context<> ctx{content_type};
     resp_data data{};
 
+    out_buf_ = {};
+
     std::shared_ptr<void> guard(nullptr, [&, this](auto) {
       if (!req_headers_.empty()) {
         req_headers_.clear();

src/coro_http/tests/test_cinatra.cpp

@@ -885,6 +885,85 @@ TEST_CASE("test pipeline") {
 }
 #endif
 
+enum class upload_type { send_file, chunked, multipart };
+
+TEST_CASE("test out buffer and async upload ") {
+  coro_http_server server(1, 9000);
+  server.set_http_handler<GET, POST>(
+      "/write_chunked",
+      [](coro_http_request &req,
+         coro_http_response &resp) -> async_simple::coro::Lazy<void> {
+        resp.set_format_type(format_type::chunked);
+        bool ok;
+        if (ok = co_await resp.get_conn()->begin_chunked(); !ok) {
+          co_return;
+        }
+
+        std::vector<std::string> vec{"hello", " world", " ok"};
+
+        for (auto &str : vec) {
+          if (ok = co_await resp.get_conn()->write_chunked(str); !ok) {
+            co_return;
+          }
+        }
+
+        ok = co_await resp.get_conn()->end_chunked();
+      });
+  server.set_http_handler<GET, POST>(
+      "/normal", [](coro_http_request &req, coro_http_response &resp) {
+        resp.set_status_and_content(status_type::ok, "test");
+      });
+  server.set_http_handler<GET, POST>(
+      "/more", [](coro_http_request &req, coro_http_response &resp) {
+        resp.set_status_and_content(status_type::ok, "test more");
+      });
+
+  server.async_start();
+
+  auto lazy = [](upload_type flag) -> async_simple::coro::Lazy<void> {
+    coro_http_client client{};
+    std::string uri = "http://127.0.0.1:9000/normal";
+    std::vector<char> oubuf;
+    oubuf.resize(10);
+    req_context<> ctx{};
+    auto result = co_await client.async_request(uri, http_method::GET,
+                                                std::move(ctx), {}, oubuf);
+    std::cout << oubuf.data() << "\n";
+
+    std::string_view out_view(oubuf.data(), result.resp_body.size());
+    assert(out_view == "test");
+    assert(out_view == result.resp_body);
+
+    auto ss = std::make_shared<std::stringstream>();
+    *ss << "hello world";
+
+    if (flag == upload_type::send_file) {
+      result = co_await client.async_upload("http://127.0.0.1:9000/more"sv,
+                                            http_method::POST, ss);
+    }
+    else if (flag == upload_type::chunked) {
+      result = co_await client.async_upload_chunked(
+          "http://127.0.0.1:9000/more"sv, http_method::POST, ss);
+    }
+    else if (flag == upload_type::multipart) {
+      client.add_str_part("test_key", "test_value");
+      result =
+          co_await client.async_upload_multipart("http://127.0.0.1:9000/more");
+    }
+
+    std::cout << (int)flag << oubuf.data() << "\n";
+    std::cout << result.resp_body << "\n";
+
+    std::string_view out_view1(oubuf.data(), out_view.size());
+    assert(out_view == out_view1);
+    assert(result.resp_body != out_view1);
+  };
+
+  async_simple::coro::syncAwait(lazy(upload_type::send_file));
+  async_simple::coro::syncAwait(lazy(upload_type::chunked));
+  async_simple::coro::syncAwait(lazy(upload_type::multipart));
+}
+
 TEST_CASE("test multipart and chunked return error") {
   coro_http_server server(1, 8090);
   server.set_http_handler<cinatra::PUT, cinatra::POST>(