Applied platform thread to network lib

PiperOrigin-RevId: 528521852
2026-09-16 15:36:12 -04:00 · 2023-05-01 11:13:36 -07:00
parent 9f6e7f6b84
commit 465aa7a489
8 changed files with 107 additions and 194 deletions
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 Google LLC
+// Copyright 2021 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -14,135 +14,107 @@

 #include "internal/network/http_client_impl.h"

+#include <chrono>  // NOLINT
 #include <functional>
-#include <memory>
+#include <future>  // NOLINT
 #include <ostream>
 #include <sstream>
 #include <utility>

-#include "absl/status/statusor.h"
 #include "internal/network/debug.h"
 #include "internal/platform/implementation/platform.h"
 #include "internal/platform/logging.h"
-#include "internal/platform/mutex_lock.h"

 namespace nearby {
 namespace network {
-namespace {
-
-// In nearby SDK, allowed maximum thread count.
-constexpr int kMaxNetworkThreadCount = 3;
-
-}  // namespace
-
-NearbyHttpClient::NearbyHttpClient() {
-  network_executor_ =
-      std::make_unique<MultiThreadExecutor>(kMaxNetworkThreadCount);
-}

 void NearbyHttpClient::StartRequest(
    const HttpRequest& request,
    std::function<void(const absl::StatusOr<HttpResponse>&)> callback) {
-  MutexLock lock(&mutex_);
-  NEARBY_LOGS(INFO) << __func__ << ": Start async request to url="
-                    << request.GetUrl().GetUrlPath();
-  if (network_executor_ == nullptr) {
-    callback(absl::ResourceExhaustedError("no available thread"));
-    return;
-  }
+  absl::MutexLock lock(&mutex_);
+  CleanThreads();

-  network_executor_->Execute([&, request, callback]() {
-    absl::StatusOr<HttpResponse> response = InternalGetResponse(request);
-    if (response.ok()) {
-      NEARBY_LOGS(INFO) << __func__ << ": Got response from url="
-                        << request.GetUrl().GetUrlPath();
-    } else {
-      NEARBY_LOGS(ERROR) << __func__ << ": Failed to get response from url="
-                         << request.GetUrl().GetUrlPath() << ", status"
-                         << response.status();
+  std::future<void> http_thread = std::async(std::launch::async, [&, request,
+                                                                  callback]() {
+    api::WebRequest web_request;
+    web_request.url = request.GetUrl().GetUrlPath();
+    web_request.method = absl::StrCat(request.GetMethodString());
+    for (const auto& header : request.GetAllHeaders()) {
+      for (const auto& value : header.second) {
+        web_request.headers.emplace(header.first, value);
+      }
+    }
+    web_request.body = absl::StrCat(request.GetBody().GetRawData());
+
+    if (debug::kRequestEnabled) {
+      std::stringstream request_stream;
+      request_stream << "HTTP REQUEST====>" << std::endl;
+      request_stream << web_request.method << " " << web_request.url
+                     << std::endl;
+      for (const auto& header : web_request.headers) {
+        request_stream << header.first << ": " << header.second << std::endl;
+      }
+      request_stream << std::endl;
+      request_stream << "body size: " << request.GetBody().GetRawData().size()
+                     << std::endl;
+      NEARBY_LOGS(VERBOSE) << request_stream.str();
    }

-    callback(response);
-    NEARBY_LOGS(INFO) << __func__ << ": Completed request to url="
-                      << request.GetUrl().GetUrlPath();
-  });
-}
+    absl::StatusOr<api::WebResponse> web_response =
+        api::ImplementationPlatform::SendRequest(web_request);

-absl::StatusOr<HttpResponse> NearbyHttpClient::GetResponse(
-    const HttpRequest& request) {
-  NEARBY_LOGS(INFO) << __func__ << ": Start request to url="
-                    << request.GetUrl().GetUrlPath();
-
-  absl::StatusOr<HttpResponse> response = InternalGetResponse(request);
-  if (response.ok()) {
-    NEARBY_LOGS(INFO) << __func__ << ": Got response from url="
-                      << request.GetUrl().GetUrlPath();
-  } else {
-    NEARBY_LOGS(ERROR) << __func__ << ": Failed to get response from url="
-                       << request.GetUrl().GetUrlPath() << ", status"
-                       << response.status();
-  }
-
-  return response;
-}
-
-absl::StatusOr<HttpResponse> NearbyHttpClient::InternalGetResponse(
-    const HttpRequest& request) {
-  api::WebRequest web_request;
-  web_request.url = request.GetUrl().GetUrlPath();
-  web_request.method = absl::StrCat(request.GetMethodString());
-  for (const auto& header : request.GetAllHeaders()) {
-    for (const auto& value : header.second) {
-      web_request.headers.emplace(header.first, value);
+    if (!web_response.ok()) {
+      if (callback != nullptr) {
+        callback(web_response.status());
+      }
+      return;
    }
-  }
-  web_request.body = absl::StrCat(request.GetBody().GetRawData());

-  if (debug::kRequestEnabled) {
-    std::stringstream request_stream;
-    request_stream << "HTTP REQUEST====>" << std::endl;
-    request_stream << web_request.method << " " << web_request.url << std::endl;
-    for (const auto& header : web_request.headers) {
-      request_stream << header.first << ": " << header.second << std::endl;
+    if (debug::kResponseEnabled) {
+      std::stringstream response_stream;
+      response_stream << "HTTP RESPONSE====>" << std::endl;
+      response_stream << "url: " << web_request.url << std::endl;
+      response_stream << web_response->status_code << " "
+                      << web_response->status_text << std::endl;
+      for (const auto& header : web_response->headers) {
+        response_stream << header.first << ": " << header.second << std::endl;
+      }
+      response_stream << std::endl;
+      response_stream << "body size: " << web_response->body.size()
+                      << std::endl;
+      NEARBY_LOGS(VERBOSE) << response_stream.str();
    }
-    request_stream << std::endl;
-    request_stream << "body size: " << request.GetBody().GetRawData().size()
-                   << std::endl;
-    NEARBY_LOGS(VERBOSE) << request_stream.str();
-  }

-  absl::StatusOr<api::WebResponse> web_response =
-      api::ImplementationPlatform::SendRequest(web_request);
+    HttpResponse response;

-  if (!web_response.ok()) {
-    return web_response.status();
-  }
-
-  if (debug::kResponseEnabled) {
-    std::stringstream response_stream;
-    response_stream << "HTTP RESPONSE====>" << std::endl;
-    response_stream << "url: " << web_request.url << std::endl;
-    response_stream << web_response->status_code << " "
-                    << web_response->status_text << std::endl;
+    response.SetStatusCode(
+        static_cast<HttpStatusCode>(web_response->status_code));
+    response.SetReasonPhrase(web_response->status_text);
    for (const auto& header : web_response->headers) {
-      response_stream << header.first << ": " << header.second << std::endl;
+      response.AddHeader(header.first, header.second);
+    }
+    response.SetBody(web_response->body);
+
+    if (callback != nullptr) {
+      callback(response);
+    }
+  });
+
+  http_threads_.push_back(std::move(http_thread));
+}
+
+void NearbyHttpClient::CleanThreads() {
+  auto it = http_threads_.begin();
+
+  while (it != http_threads_.end()) {
+    // Delete the thread if it is ready
+    auto status = it->wait_for(std::chrono::seconds(0));
+    if (status == std::future_status::ready) {
+      it = http_threads_.erase(it);
+    } else {
+      ++it;
    }
-    response_stream << std::endl;
-    response_stream << "body size: " << web_response->body.size() << std::endl;
-    NEARBY_LOGS(VERBOSE) << response_stream.str();
  }
-
-  HttpResponse response;
-
-  response.SetStatusCode(
-      static_cast<HttpStatusCode>(web_response->status_code));
-  response.SetReasonPhrase(web_response->status_text);
-  for (const auto& header : web_response->headers) {
-    response.AddHeader(header.first, header.second);
-  }
-  response.SetBody(web_response->body);
-
-  return response;
 }

 }  // namespace network