TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 22:56:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Applied platform thread to network lib

Browse Source 
PiperOrigin-RevId: 520998502
This commit is contained in:
Guogang Li
2023-03-31 13:23:32 -07:00
committed by Copybara-Service
parent 3e2d4b09ff
commit be99b60feb
8 changed files with 198 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastpair/internal/test/BUILD
+3
View File
@@ -14,6 +14,8 @@ cc_library(
deps = [
"//internal/network:types",
"@com_google_absl//absl/functional:any_invocable",
"@com_google_absl//absl/status",
"@com_google_absl//absl/status:statusor",
],
)
@@ -30,6 +32,7 @@ cc_test(
deps = [
":nearby_fastpair_test",
"//internal/network:types",
"//internal/platform/implementation/g3",
"@com_github_protobuf_matchers//protobuf-matchers",
"@com_google_absl//absl/container:flat_hash_map",
"@com_google_absl//absl/functional:any_invocable",
fastpair/internal/test/fast_pair_fake_http_client.h
+6
View File
@@ -21,6 +21,7 @@
#include <utility>
#include <vector>
#include "absl/status/statusor.h"
#include "internal/network/http_client.h"
namespace nearby {
@@ -51,6 +52,11 @@ class FastPairFakeHttpClient : public HttpClient {
request_infos_.push_back(std::move(request_info));
}
absl::StatusOr<HttpResponse> GetResponse(
const HttpRequest& request) override {
return absl::UnimplementedError("unimplemented");
}
// Mock methods
void CompleteRequest(const absl::StatusOr<HttpResponse>& response,
size_t pos = 0) {
internal/network/BUILD
+3 -2
View File
@@ -31,6 +31,7 @@ cc_library(
"//presence:__subpackages__",
],
deps = [
"//internal/platform:types",
"@com_google_absl//absl/container:flat_hash_map",
"@com_google_absl//absl/status",
"@com_google_absl//absl/status:statusor",
@@ -63,11 +64,11 @@ cc_library(
deps = [
":types",
"//internal/platform:logging",
"//internal/platform:types",
"//internal/platform/implementation:platform",
"@com_google_absl//absl/base:core_headers",
"@com_google_absl//absl/container:flat_hash_map",
"@com_google_absl//absl/status:statusor",
"@com_google_absl//absl/strings:str_format",
"@com_google_absl//absl/synchronization",
],
)
internal/network/http_client.h
+6 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2021 Google LLC
// Copyright 2021-2023 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -28,10 +28,15 @@ class HttpClient {
public:
virtual ~HttpClient() = default;
// Starts HTTP request in asynchronization mode.
virtual void StartRequest(
const HttpRequest& request,
std::function<void(const absl::StatusOr<HttpResponse>&)> callback) = 0;
// Gets HTTP response in synchronization mode.
virtual absl::StatusOr<HttpResponse> GetResponse(
const HttpRequest& request) = 0;
// The error may be corrected if retried at a later time.
static bool IsRetryableHttpError(absl::Status status) {
return absl::IsUnavailable(status) || absl::IsFailedPrecondition(status);
internal/network/http_client_impl.cc
+106 -78
View File
@@ -1,4 +1,4 @@
// Copyright 2021 Google LLC
// Copyright 2021-2023 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,107 +14,135 @@
#include "internal/network/http_client_impl.h"
#include <chrono> // NOLINT
#include <functional>
#include <future> // NOLINT
#include <memory>
#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) {
absl::MutexLock lock(&mutex_);
CleanThreads();
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;
}
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();
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();
}
absl::StatusOr<api::WebResponse> web_response =
api::ImplementationPlatform::SendRequest(web_request);
if (!web_response.ok()) {
if (callback != nullptr) {
callback(web_response.status());
}
return;
}
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();
}
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);
if (callback != nullptr) {
callback(response);
}
callback(response);
NEARBY_LOGS(INFO) << __func__ << ": Completed request to url="
<< request.GetUrl().GetUrlPath();
});
http_threads_.push_back(std::move(http_thread));
}
void NearbyHttpClient::CleanThreads() {
auto it = http_threads_.begin();
absl::StatusOr<HttpResponse> NearbyHttpClient::GetResponse(
const HttpRequest& request) {
NEARBY_LOGS(INFO) << __func__ << ": Start request to url="
<< request.GetUrl().GetUrlPath();
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;
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);
}
}
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();
}
absl::StatusOr<api::WebResponse> web_response =
api::ImplementationPlatform::SendRequest(web_request);
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;
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();
}
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
internal/network/http_client_impl.h
+14 -11
View File
@@ -1,4 +1,4 @@
// Copyright 2022 Google LLC
// Copyright 2022-2023 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -16,21 +16,20 @@
#define THIRD_PARTY_NEARBY_INTERNAL_NETWORK_HTTP_CLIENT_IMPL_H_
#include <functional>
#include <future> // NOLINT
#include <thread> // NOLINT
#include <utility>
#include <vector>
#include <memory>
#include "absl/base/thread_annotations.h"
#include "absl/synchronization/mutex.h"
#include "absl/status/statusor.h"
#include "internal/network/http_client.h"
#include "internal/platform/multi_thread_executor.h"
#include "internal/platform/mutex.h"
namespace nearby {
namespace network {
class NearbyHttpClient : public HttpClient {
public:
NearbyHttpClient() = default;
NearbyHttpClient();
~NearbyHttpClient() override = default;
NearbyHttpClient(const NearbyHttpClient&) = default;
@@ -38,15 +37,19 @@ class NearbyHttpClient : public HttpClient {
NearbyHttpClient(NearbyHttpClient&&) = default;
NearbyHttpClient& operator=(NearbyHttpClient&&) = default;
// Starts HTTP request in asynchronization mode.
void StartRequest(const HttpRequest& request,
std::function<void(const absl::StatusOr<HttpResponse>&)>
callback) override ABSL_LOCKS_EXCLUDED(mutex_);
private:
void CleanThreads() ABSL_SHARED_LOCKS_REQUIRED(mutex_);
// Gets HTTP response in synchronization mode.
absl::StatusOr<HttpResponse> GetResponse(const HttpRequest& request) override;
absl::Mutex mutex_;
std::vector<std::future<void>> http_threads_ ABSL_GUARDED_BY(mutex_);
private:
absl::StatusOr<HttpResponse> InternalGetResponse(const HttpRequest& request);
Mutex mutex_;
std::unique_ptr<MultiThreadExecutor> network_executor_ = nullptr;
};
} // namespace network
internal/network/http_client_impl_test.cc
+59 -19
View File
@@ -1,4 +1,4 @@
// Copyright 2021 Google LLC
// Copyright 2021-2023 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -89,7 +89,7 @@ class NearbyHttpClientTest : public ::testing::Test {
api::WebRequest GetWebRequest() { return api::GetContext()->web_request; }
absl::StatusOr<HttpResponse> GetResponse(
absl::StatusOr<HttpResponse> GetResponseAsync(
absl::string_view url, HttpRequestMethod method,
const std::multimap<std::string, std::string>& headers,
absl::string_view body) {
@@ -121,6 +121,28 @@ class NearbyHttpClientTest : public ::testing::Test {
return result;
}
absl::StatusOr<HttpResponse> GetResponse(
absl::string_view url, HttpRequestMethod method,
const std::multimap<std::string, std::string>& headers,
absl::string_view body) {
absl::StatusOr<HttpResponse> result;
absl::StatusOr<Url> request_url = Url::Create(url);
if (!request_url.ok()) {
return request_url.status();
}
HttpRequest request{request_url.value()};
auto it = headers.begin();
while (it != headers.end()) {
request.AddHeader(it->first, it->second);
++it;
}
request.SetMethod(method);
request.SetBody(body);
return client_.GetResponse(request);
}
void CheckHeader(const std::multimap<std::string, std::string>& headers,
absl::string_view key, absl::string_view expected_value) {
auto it = headers.find(std::string(key));
@@ -146,8 +168,8 @@ namespace {
TEST_F(NearbyHttpClientTest, TestGet) {
MockResponse(HttpStatusCode::kHttpOk, "OK", {{"Content_Type", "text/html"}},
"web content");
auto result =
GetResponse("http://www.google.com", HttpRequestMethod::kGet, {}, "");
auto result = GetResponseAsync("http://www.google.com",
HttpRequestMethod::kGet, {}, "");
// Checks request.
api::WebRequest web_request = GetWebRequest();
@@ -165,8 +187,8 @@ TEST_F(NearbyHttpClientTest, TestGet) {
TEST_F(NearbyHttpClientTest, TestGetWithQuery) {
MockResponse(HttpStatusCode::kHttpOk, "OK", {{"Content_Type", "text/html"}},
"web content");
auto result = GetResponse("http://www.google.com?name=name1&age=36",
HttpRequestMethod::kGet, {}, "");
auto result = GetResponseAsync("http://www.google.com?name=name1&age=36",
HttpRequestMethod::kGet, {}, "");
// Checks request.
api::WebRequest web_request = GetWebRequest();
@@ -180,8 +202,8 @@ TEST_F(NearbyHttpClientTest, TestGetWithQuery) {
TEST_F(NearbyHttpClientTest, TestGetWithErrorResult) {
MockFailedResponse(absl::InternalError("no connection."));
auto result = GetResponse("http://www.google.com?name=name1&age=36",
HttpRequestMethod::kGet, {}, "");
auto result = GetResponseAsync("http://www.google.com?name=name1&age=36",
HttpRequestMethod::kGet, {}, "");
// Checks request.
api::WebRequest web_request = GetWebRequest();
@@ -192,6 +214,24 @@ TEST_F(NearbyHttpClientTest, TestGetWithErrorResult) {
EXPECT_FALSE(result.ok());
}
TEST_F(NearbyHttpClientTest, TestPostAsync) {
MockResponse(HttpStatusCode::kHttpNoContent, "OK",
{{"Content_Type", "text/html"}}, "");
auto result = GetResponseAsync("http://www.google.com",
HttpRequestMethod::kPost, {}, "");
// Checks request.
api::WebRequest web_request = GetWebRequest();
EXPECT_EQ(web_request.url, "http://www.google.com");
EXPECT_EQ(web_request.method, "POST");
// Checks response.
ASSERT_TRUE(result.ok());
EXPECT_EQ(result->GetStatusCode(), HttpStatusCode::kHttpNoContent);
HttpResponseBody body = result->GetBody();
EXPECT_TRUE(body.empty());
}
TEST_F(NearbyHttpClientTest, TestPost) {
MockResponse(HttpStatusCode::kHttpNoContent, "OK",
{{"Content_Type", "text/html"}}, "");
@@ -210,12 +250,12 @@ TEST_F(NearbyHttpClientTest, TestPost) {
EXPECT_TRUE(body.empty());
}
TEST_F(NearbyHttpClientTest, TestPostWithHeader) {
TEST_F(NearbyHttpClientTest, TestPostWithHeaderAsync) {
MockResponse(HttpStatusCode::kHttpNoContent, "OK",
{{"Content_Type", "text/html"}}, "");
auto result =
GetResponse("http://www.google.com", HttpRequestMethod::kPost,
{{"Content_Type", "text/json"}, {"size", "596"}}, "");
GetResponseAsync("http://www.google.com", HttpRequestMethod::kPost,
{{"Content_Type", "text/json"}, {"size", "596"}}, "");
// Checks request.
api::WebRequest web_request = GetWebRequest();
@@ -231,11 +271,11 @@ TEST_F(NearbyHttpClientTest, TestPostWithHeader) {
EXPECT_EQ(result->GetBody().GetRawData(), "");
}
TEST_F(NearbyHttpClientTest, TestPostWithErrorResult) {
TEST_F(NearbyHttpClientTest, TestPostWithErrorResultAsync) {
MockFailedResponse(absl::UnauthenticatedError("no user."));
auto result =
GetResponse("http://www.google.com", HttpRequestMethod::kPost,
{{"Content_Type", "text/json"}, {"size", "596"}}, "");
GetResponseAsync("http://www.google.com", HttpRequestMethod::kPost,
{{"Content_Type", "text/json"}, {"size", "596"}}, "");
// Checks request.
api::WebRequest web_request = GetWebRequest();
@@ -249,11 +289,11 @@ TEST_F(NearbyHttpClientTest, TestPostWithErrorResult) {
ASSERT_FALSE(result.ok());
}
TEST_F(NearbyHttpClientTest, TestRequestWithCleanThreads) {
TEST_F(NearbyHttpClientTest, TestRequestWithCleanThreadsAsync) {
MockResponse(HttpStatusCode::kHttpOk, "OK", {{"Content_Type", "text/html"}},
"web content");
auto result =
GetResponse("http://www.google.com", HttpRequestMethod::kGet, {}, "");
auto result = GetResponseAsync("http://www.google.com",
HttpRequestMethod::kGet, {}, "");
// Checks request.
api::WebRequest web_request = GetWebRequest();
@@ -263,8 +303,8 @@ TEST_F(NearbyHttpClientTest, TestRequestWithCleanThreads) {
// Checks response.
ASSERT_TRUE(result.ok());
result =
GetResponse("http://www.youtube.com", HttpRequestMethod::kGet, {}, "");
result = GetResponseAsync("http://www.youtube.com", HttpRequestMethod::kGet,
{}, "");
ASSERT_TRUE(result.ok());
}
internal/platform/BUILD
+1
View File
@@ -342,6 +342,7 @@ cc_library(
"//fastpair:__subpackages__",
"//internal/base:__subpackages__",
"//internal/flags:__subpackages__",
"//internal/network:__subpackages__",
"//internal/platform/implementation/windows:__subpackages__",
"//internal/test:__subpackages__",
"//internal/weave:__pkg__",