Run CURL in background thread. BUG:24204632 Change-Id: I7f62ccec837ef07be47a52c426b1a27f72f55ac1 Reviewed-on: https://weave-review.googlesource.com/1483 Reviewed-by: Vitaly Buka <vitalybuka@google.com>

diff --git a/examples/provider/curl_http_client.cc b/examples/provider/curl_http_client.cc
index b440f39..3dcfe3d 100644
--- a/examples/provider/curl_http_client.cc
+++ b/examples/provider/curl_http_client.cc

@@ -4,6 +4,9 @@
 
 #include "examples/provider/curl_http_client.h"
 
+#include <future>
+#include <thread>
+
 #include <base/bind.h>
 #include <curl/curl.h>
 #include <weave/provider/task_runner.h>
@@ -30,29 +33,24 @@
   return size * nmemb;
 }
 
-}  // namespace
-
-CurlHttpClient::CurlHttpClient(provider::TaskRunner* task_runner)
-    : task_runner_{task_runner} {}
-
-void CurlHttpClient::SendRequest(Method method,
-                                 const std::string& url,
-                                 const Headers& headers,
-                                 const std::string& data,
-                                 const SendRequestCallback& callback) {
+std::pair<std::unique_ptr<CurlHttpClient::Response>, ErrorPtr>
+SendRequestBlocking(CurlHttpClient::Method method,
+                    const std::string& url,
+                    const CurlHttpClient::Headers& headers,
+                    const std::string& data) {
   std::unique_ptr<CURL, decltype(&curl_easy_cleanup)> curl{curl_easy_init(),
                                                            &curl_easy_cleanup};
   CHECK(curl);
 
   switch (method) {
-    case Method::kGet:
+    case CurlHttpClient::Method::kGet:
       CHECK_EQ(CURLE_OK, curl_easy_setopt(curl.get(), CURLOPT_HTTPGET, 1L));
       break;
-    case Method::kPost:
+    case CurlHttpClient::Method::kPost:
       CHECK_EQ(CURLE_OK, curl_easy_setopt(curl.get(), CURLOPT_HTTPPOST, 1L));
       break;
-    case Method::kPatch:
-    case Method::kPut:
+    case CurlHttpClient::Method::kPatch:
+    case CurlHttpClient::Method::kPut:
       CHECK_EQ(CURLE_OK, curl_easy_setopt(curl.get(), CURLOPT_CUSTOMREQUEST,
                                           weave::EnumToString(method).c_str()));
       break;
@@ -66,7 +64,7 @@
 
   CHECK_EQ(CURLE_OK, curl_easy_setopt(curl.get(), CURLOPT_HTTPHEADER, chunk));
 
-  if (!data.empty() || method == Method::kPost) {
+  if (!data.empty() || method == CurlHttpClient::Method::kPost) {
     CHECK_EQ(CURLE_OK,
              curl_easy_setopt(curl.get(), CURLOPT_POSTFIELDS, data.c_str()));
   }
@@ -89,8 +87,7 @@
   if (res != CURLE_OK) {
     Error::AddTo(&error, FROM_HERE, "curl", "curl_easy_perform_error",
                  curl_easy_strerror(res));
-    return task_runner_->PostDelayedTask(
-        FROM_HERE, base::Bind(callback, nullptr, base::Passed(&error)), {});
+    return {nullptr, std::move(error)};
   }
 
   const std::string kContentType = "\r\nContent-Type:";
@@ -98,8 +95,7 @@
   if (pos == std::string::npos) {
     Error::AddTo(&error, FROM_HERE, "curl", "no_content_header",
                  "Content-Type header is missing");
-    return task_runner_->PostDelayedTask(
-        FROM_HERE, base::Bind(callback, nullptr, base::Passed(&error)), {});
+    return {nullptr, std::move(error)};
   }
   pos += kContentType.size();
   auto pos_end = response->content_type.find("\r\n", pos);
@@ -112,8 +108,56 @@
   CHECK_EQ(CURLE_OK, curl_easy_getinfo(curl.get(), CURLINFO_RESPONSE_CODE,
                                        &response->status));
 
+  return {std::move(response), nullptr};
+}
+
+}  // namespace
+
+CurlHttpClient::CurlHttpClient(provider::TaskRunner* task_runner)
+    : task_runner_{task_runner} {}
+
+void CurlHttpClient::SendRequest(Method method,
+                                 const std::string& url,
+                                 const Headers& headers,
+                                 const std::string& data,
+                                 const SendRequestCallback& callback) {
+  pending_tasks_.emplace_back(
+      std::async(std::launch::async, SendRequestBlocking, method, url, headers,
+                 data),
+      callback);
+  if (pending_tasks_.size() == 1)  // More means check is scheduled.
+    CheckTasks();
+}
+
+void CurlHttpClient::CheckTasks() {
+  VLOG(3) << "CurlHttpClient::CheckTasks, size=" << pending_tasks_.size();
+  auto ready_begin =
+      std::remove_if(pending_tasks_.begin(), pending_tasks_.end(),
+                     [](const decltype(pending_tasks_)::value_type& value) {
+                       return value.first.wait_for(std::chrono::seconds(0)) !=
+                              std::future_status::ready;
+                     });
+
+  for (auto it = ready_begin; it != pending_tasks_.end(); ++it) {
+    auto result = it->first.get();
+    VLOG(2) << "CurlHttpClient::CheckTasks done";
+    task_runner_->PostDelayedTask(
+        FROM_HERE, base::Bind(it->second, base::Passed(&result.first),
+                              base::Passed(&result.second)),
+        {});
+  }
+
+  pending_tasks_.erase(ready_begin, pending_tasks_.end());
+
+  if (pending_tasks_.empty()) {
+    VLOG(2) << "No more CurlHttpClient tasks";
+    return;
+  }
+
   task_runner_->PostDelayedTask(
-      FROM_HERE, base::Bind(callback, base::Passed(&response), nullptr), {});
+      FROM_HERE,
+      base::Bind(&CurlHttpClient::CheckTasks, weak_ptr_factory_.GetWeakPtr()),
+      base::TimeDelta::FromMilliseconds(100));
 }
 
 }  // namespace examples

diff --git a/examples/provider/curl_http_client.h b/examples/provider/curl_http_client.h
index b2f4bca..c342076 100644
--- a/examples/provider/curl_http_client.h
+++ b/examples/provider/curl_http_client.h

@@ -5,7 +5,9 @@
 #ifndef LIBWEAVE_EXAMPLES_PROVIDER_CURL_HTTP_CLIENT_H_
 #define LIBWEAVE_EXAMPLES_PROVIDER_CURL_HTTP_CLIENT_H_
 
+#include <future>
 #include <string>
+#include <utility>
 
 #include <base/memory/weak_ptr.h>
 #include <weave/provider/http_client.h>
@@ -31,6 +33,11 @@
                    const SendRequestCallback& callback) override;
 
  private:
+  void CheckTasks();
+
+  std::vector<
+      std::pair<std::future<std::pair<std::unique_ptr<Response>, ErrorPtr>>,
+                SendRequestCallback>> pending_tasks_;
   provider::TaskRunner* task_runner_{nullptr};
 
   base::WeakPtrFactory<CurlHttpClient> weak_ptr_factory_{this};