BlockStorage/repertory
1
0
Fork 0
Code Issues 3 Pull Requests Releases 9 Wiki Activity

v2.0.1-rc (#13)

Browse Source 
Reviewed-on: #13
This commit is contained in:
sgraves76
2023-12-10 18:11:20 +00:00
parent dc410bfe5b
commit 40f6054dcc
404 changed files with 410072 additions and 146106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

366
src/comm/curl/curl_comm.cpp
View File

@@ -1,176 +1,190 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/curl/curl_comm.hpp"
#include "types/repertory.hpp"
#include "utils/path_utils.hpp"
#include "utils/string_utils.hpp"
namespace repertory {
const curl_comm::write_callback curl_comm::write_data =
static_cast<curl_comm::write_callback>([](char *buffer, size_t size,
size_t nitems,
void *outstream) -> size_t {
auto &info = *reinterpret_cast<read_write_info *>(outstream);
std::copy(buffer, buffer + (size * nitems),
std::back_inserter(info.data));
return info.stop_requested ? 0 : size * nitems;
});
const curl_comm::write_callback curl_comm::write_headers =
static_cast<curl_comm::write_callback>([](char *buffer, size_t size,
size_t nitems,
void *outstream) -> size_t {
auto &headers = *reinterpret_cast<http_headers *>(outstream);
const auto header = std::string(buffer, size * nitems);
const auto parts = utils::string::split(header, ':');
if (parts.size() > 1u) {
auto data = header.substr(parts[0u].size() + 1u);
utils::string::left_trim(data);
utils::string::right_trim(data, '\r');
utils::string::right_trim(data, '\n');
utils::string::right_trim(data, '\r');
headers[utils::string::to_lower(parts[0u])] = data;
}
return size * nitems;
});
curl_comm::curl_comm(host_config hc)
: host_config_(std::move(hc)), s3_config_(std::nullopt) {}
curl_comm::curl_comm(s3_config s3)
: host_config_(std::nullopt), s3_config_(std::move(s3)) {}
auto curl_comm::construct_url(CURL *curl, const std::string &relative_path,
const host_config &hc) -> std::string {
auto custom_port =
(((hc.protocol == "http") && (hc.api_port == 80U || hc.api_port == 0U)) ||
((hc.protocol == "https") && (hc.api_port == 443U || hc.api_port == 0U)))
? ""
: ":" + std::to_string(hc.api_port);
auto url = hc.protocol + "://" +
utils::string::trim_copy(hc.host_name_or_ip) + custom_port;
static const auto complete_url = [](const std::string &current_path,
const std::string &parent_path,
std::string &final_url) -> std::string & {
final_url += utils::path::create_api_path(current_path);
if (utils::string::ends_with(parent_path, "/")) {
final_url += '/';
}
return final_url;
};
auto path = utils::path::combine("/", {hc.path});
return relative_path.empty()
? complete_url(path, hc.path, url)
: complete_url(utils::path::combine(
path, {url_encode(curl, relative_path, true)}),
relative_path, url);
}
auto curl_comm::create_host_config(const s3_config &config,
bool use_s3_path_style) -> host_config {
host_config hc{};
hc.api_password = config.secret_key;
hc.api_user = config.access_key;
auto pos = config.url.find(':');
hc.host_name_or_ip = config.url.substr(pos + 3U);
if (config.use_region_in_url && not config.region.empty()) {
auto parts = utils::string::split(hc.host_name_or_ip, '.', false);
if (parts.size() > 1U) {
parts.insert(parts.begin() + 1U, config.region);
hc.host_name_or_ip = utils::string::join(parts, '.');
}
}
if (not use_s3_path_style) {
hc.host_name_or_ip = config.bucket + '.' + hc.host_name_or_ip;
}
hc.protocol = config.url.substr(0U, pos);
if (use_s3_path_style) {
hc.path = '/' + config.bucket;
}
return hc;
}
void curl_comm::enable_s3_path_style(bool enable) {
use_s3_path_style_ = enable;
}
auto curl_comm::make_request(const curl::requests::http_delete &del,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value(),
del, response_code, stop_requested);
}
auto curl_comm::make_request(const curl::requests::http_get &get,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value(),
get, response_code, stop_requested);
}
auto curl_comm::make_request(const curl::requests::http_head &head,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value(),
head, response_code, stop_requested);
}
auto curl_comm::make_request(const curl::requests::http_put_file &put_file,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value(),
put_file, response_code, stop_requested);
}
auto curl_comm::url_encode(CURL *curl, const std::string &data,
bool allow_slash) -> std::string {
auto *value =
curl_easy_escape(curl, data.c_str(), static_cast<int>(data.size()));
std::string ret = value;
curl_free(value);
if (allow_slash) {
utils::string::replace(ret, "%2F", "/");
}
return ret;
}
} // namespace repertory
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/curl/curl_comm.hpp"
#include "types/repertory.hpp"
#include "utils/path_utils.hpp"
#include "utils/string_utils.hpp"
namespace repertory {
const curl_comm::write_callback curl_comm::write_data =
static_cast<curl_comm::write_callback>([](char *buffer, size_t size,
size_t nitems,
void *outstream) -> size_t {
auto &info = *reinterpret_cast<read_write_info *>(outstream);
std::copy(buffer, buffer + (size * nitems),
std::back_inserter(info.data));
return info.stop_requested ? 0 : size * nitems;
});
const curl_comm::write_callback curl_comm::write_headers =
static_cast<curl_comm::write_callback>([](char *buffer, size_t size,
size_t nitems,
void *outstream) -> size_t {
auto &headers = *reinterpret_cast<http_headers *>(outstream);
const auto header = std::string(buffer, size * nitems);
const auto parts = utils::string::split(header, ':');
if (parts.size() > 1U) {
auto data = header.substr(parts[0U].size() + 1U);
utils::string::left_trim(data);
utils::string::right_trim(data, '\r');
utils::string::right_trim(data, '\n');
utils::string::right_trim(data, '\r');
headers[utils::string::to_lower(parts[0U])] = data;
}
return size * nitems;
});
curl_comm::curl_comm(host_config cfg)
: host_config_(std::move(cfg)), s3_config_(std::nullopt) {}
curl_comm::curl_comm(s3_config cfg)
: host_config_(std::nullopt), s3_config_(std::move(cfg)) {}
auto curl_comm::construct_url(CURL *curl, const std::string &relative_path,
const host_config &cfg) -> std::string {
static constexpr const auto http = 80U;
static constexpr const auto https = 443U;
auto custom_port = (((cfg.protocol == "http") &&
(cfg.api_port == http || cfg.api_port == 0U)) ||
((cfg.protocol == "https") &&
(cfg.api_port == https || cfg.api_port == 0U)))
? ""
: ":" + std::to_string(cfg.api_port);
auto url = cfg.protocol + "://" +
utils::string::trim_copy(cfg.host_name_or_ip) + custom_port;
static const auto complete_url = [](const std::string &current_path,
const std::string &parent_path,
std::string &final_url) -> std::string & {
final_url += utils::path::create_api_path(current_path);
if (utils::string::ends_with(parent_path, "/")) {
final_url += '/';
}
return final_url;
};
auto path = utils::path::combine("/", {cfg.path});
return relative_path.empty()
? complete_url(path, cfg.path, url)
: complete_url(utils::path::combine(
path, {url_encode(curl, relative_path, true)}),
relative_path, url);
}
auto curl_comm::create_host_config(const s3_config &cfg, bool use_s3_path_style)
-> host_config {
host_config host_cfg{};
host_cfg.api_password = cfg.secret_key;
host_cfg.api_user = cfg.access_key;
auto pos = cfg.url.find(':');
host_cfg.host_name_or_ip = cfg.url.substr(pos + 3U);
if (cfg.use_region_in_url && not cfg.region.empty()) {
auto parts = utils::string::split(host_cfg.host_name_or_ip, '.', false);
if (parts.size() > 1U) {
parts.insert(parts.begin() + 1U, cfg.region);
host_cfg.host_name_or_ip = utils::string::join(parts, '.');
}
}
if (not use_s3_path_style) {
host_cfg.host_name_or_ip = cfg.bucket + '.' + host_cfg.host_name_or_ip;
}
host_cfg.protocol = cfg.url.substr(0U, pos);
if (use_s3_path_style) {
host_cfg.path = '/' + cfg.bucket;
}
return host_cfg;
}
void curl_comm::enable_s3_path_style(bool enable) {
use_s3_path_style_ = enable;
}
auto curl_comm::make_request(const curl::requests::http_delete &del,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value_or(host_config{}),
del, response_code, stop_requested);
}
auto curl_comm::make_request(const curl::requests::http_get &get,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value_or(host_config{}),
get, response_code, stop_requested);
}
auto curl_comm::make_request(const curl::requests::http_head &head,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value_or(host_config{}),
head, response_code, stop_requested);
}
auto curl_comm::make_request(const curl::requests::http_post &post,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value_or(host_config{}),
post, response_code, stop_requested);
}
auto curl_comm::make_request(const curl::requests::http_put_file &put_file,
long &response_code,
stop_type &stop_requested) const -> bool {
return make_request(
s3_config_.has_value()
? create_host_config(s3_config_.value(), use_s3_path_style_)
: host_config_.value_or(host_config{}),
put_file, response_code, stop_requested);
}
auto curl_comm::url_encode(CURL *curl, const std::string &data,
bool allow_slash) -> std::string {
auto *value =
curl_easy_escape(curl, data.c_str(), static_cast<int>(data.size()));
std::string ret = value;
curl_free(value);
if (allow_slash) {
utils::string::replace(ret, "%2F", "/");
}
return ret;
}
} // namespace repertory

134
src/comm/curl/multi_request.cpp
View File

@@ -1,66 +1,68 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/curl/multi_request.hpp"
#include "utils/utils.hpp"
namespace repertory {
multi_request::multi_request(CURL *curl_handle, stop_type &stop_requested)
: curl_handle_(curl_handle),
stop_requested_(stop_requested),
multi_handle_(curl_multi_init()) {
curl_multi_add_handle(multi_handle_, curl_handle);
}
multi_request::~multi_request() {
curl_multi_remove_handle(multi_handle_, curl_handle_);
curl_easy_cleanup(curl_handle_);
curl_multi_cleanup(multi_handle_);
}
void multi_request::get_result(CURLcode &curl_code, long &http_code) {
curl_code = CURLcode::CURLE_ABORTED_BY_CALLBACK;
http_code = -1;
auto error = false;
int running_handles = 0;
curl_multi_perform(multi_handle_, &running_handles);
while (not error && (running_handles > 0) && not stop_requested_) {
int ignored;
curl_multi_wait(multi_handle_, nullptr, 0, 100, &ignored);
const auto ret = curl_multi_perform(multi_handle_, &running_handles);
error = (ret != CURLM_CALL_MULTI_PERFORM) && (ret != CURLM_OK);
}
if (not stop_requested_) {
int remaining_messages = 0;
auto *multi_result =
curl_multi_info_read(multi_handle_, &remaining_messages);
if (multi_result && (multi_result->msg == CURLMSG_DONE)) {
curl_easy_getinfo(multi_result->easy_handle, CURLINFO_RESPONSE_CODE,
&http_code);
curl_code = multi_result->data.result;
}
}
}
} // namespace repertory
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/curl/multi_request.hpp"
#include "utils/utils.hpp"
namespace repertory {
multi_request::multi_request(CURL *curl_handle, stop_type &stop_requested)
: curl_handle_(curl_handle),
stop_requested_(stop_requested),
multi_handle_(curl_multi_init()) {
curl_multi_add_handle(multi_handle_, curl_handle);
}
multi_request::~multi_request() {
curl_multi_remove_handle(multi_handle_, curl_handle_);
curl_easy_cleanup(curl_handle_);
curl_multi_cleanup(multi_handle_);
}
void multi_request::get_result(CURLcode &curl_code, long &http_code) {
static constexpr const auto timeout_ms = 100;
curl_code = CURLcode::CURLE_ABORTED_BY_CALLBACK;
http_code = -1;
auto error = false;
int running_handles = 0;
curl_multi_perform(multi_handle_, &running_handles);
while (not error && (running_handles > 0) && not stop_requested_) {
int ignored{};
curl_multi_wait(multi_handle_, nullptr, 0, timeout_ms, &ignored);
const auto ret = curl_multi_perform(multi_handle_, &running_handles);
error = (ret != CURLM_CALL_MULTI_PERFORM) && (ret != CURLM_OK);
}
if (not stop_requested_) {
int remaining_messages = 0;
auto *multi_result =
curl_multi_info_read(multi_handle_, &remaining_messages);
if ((multi_result != nullptr) && (multi_result->msg == CURLMSG_DONE)) {
curl_easy_getinfo(multi_result->easy_handle, CURLINFO_RESPONSE_CODE,
&http_code);
curl_code = multi_result->data.result;
}
}
}
} // namespace repertory

45
src/comm/curl/requests/http_post.cpp Normal file
View File

@@ -0,0 +1,45 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/curl/requests/http_post.hpp"
namespace repertory::curl::requests {
http_post::~http_post() {
if (headers != nullptr) {
curl_slist_free_all(headers);
}
}
auto http_post::set_method(CURL *curl, stop_type & /*stop_requested*/) const
-> bool {
curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "POST");
if (json.has_value()) {
headers = curl_slist_append(headers, "content-type: application/json");
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
json_str = json->dump();
curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_str->c_str());
curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, -1L);
}
return true;
}
} // namespace repertory::curl::requests

149
src/comm/curl/requests/http_put_file.cpp
View File

@@ -1,80 +1,69 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/curl/requests/http_put_file.hpp"
#include "utils/string_utils.hpp"
namespace repertory::curl::requests {
auto http_put_file::get_path() const -> std::string {
if (reader) {
auto updated_path = path;
return utils::string::replace(updated_path, file_name,
reader->get_encrypted_file_name());
}
return http_request_base::get_path();
}
auto http_put_file::set_method(CURL *curl, stop_type &stop_requested) const
-> bool {
curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PUT");
if (not source_path.empty()) {
curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L);
if (not encryption_token.value_or("").empty()) {
if (not reader) {
reader = std::make_shared<utils::encryption::encrypting_reader>(
file_name, source_path, stop_requested, encryption_token.value());
}
curl_easy_setopt(curl, CURLOPT_READDATA, reader.get());
curl_easy_setopt(
curl, CURLOPT_READFUNCTION,
static_cast<curl_read_callback>(
utils::encryption::encrypting_reader::reader_function));
curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE,
reader->get_total_size());
return true;
}
read_info = std::make_shared<read_file_info>(read_file_info{
stop_requested,
});
if (native_file::open(source_path, read_info->nf) != api_error::success) {
return false;
}
read_info->nf->set_auto_close(true);
std::uint64_t file_size{};
if (not read_info->nf->get_file_size(file_size)) {
return false;
}
curl_easy_setopt(curl, CURLOPT_READDATA, read_info.get());
curl_easy_setopt(curl, CURLOPT_READFUNCTION, read_file_data);
curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, file_size);
}
return true;
}
} // namespace repertory::curl::requests
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/curl/requests/http_put_file.hpp"
#include "utils/string_utils.hpp"
namespace repertory::curl::requests {
auto http_put_file::set_method(CURL *curl, stop_type &stop_requested) const
-> bool {
curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PUT");
curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L);
if (source_path.empty()) {
curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, 0L);
return true;
}
if (reader) {
curl_easy_setopt(curl, CURLOPT_READDATA, reader.get());
curl_easy_setopt(
curl, CURLOPT_READFUNCTION,
static_cast<curl_read_callback>(
utils::encryption::encrypting_reader::reader_function));
curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, reader->get_total_size());
return true;
}
read_info = std::make_shared<read_file_info>(read_file_info{
stop_requested,
});
if (native_file::create_or_open(source_path, read_info->nf) !=
api_error::success) {
return false;
}
read_info->nf->set_auto_close(true);
std::uint64_t file_size{};
if (not read_info->nf->get_file_size(file_size)) {
return false;
}
curl_easy_setopt(curl, CURLOPT_READDATA, read_info.get());
curl_easy_setopt(curl, CURLOPT_READFUNCTION, read_file_data);
curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, file_size);
return true;
}
} // namespace repertory::curl::requests

320
src/comm/packet/client_pool.cpp
View File

@@ -1,160 +1,160 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/packet/client_pool.hpp"
#include "events/event_system.hpp"
#include "events/events.hpp"
#include "utils/error_utils.hpp"
namespace repertory {
void client_pool::pool::execute(
std::uint64_t thread_id, const worker_callback &worker,
const worker_complete_callback &worker_complete) {
const auto index = thread_id % pool_queues_.size();
auto wi = std::make_shared<work_item>(worker, worker_complete);
auto &pool_queue = pool_queues_[index];
unique_mutex_lock queue_lock(pool_queue->mutex);
pool_queue->queue.emplace_back(wi);
pool_queue->notify.notify_all();
queue_lock.unlock();
}
client_pool::pool::pool(std::uint8_t pool_size) {
event_system::instance().raise<service_started>("client_pool");
thread_index_ = 0u;
for (std::uint8_t i = 0u; i < pool_size; i++) {
pool_queues_.emplace_back(std::make_unique<work_queue>());
}
for (std::size_t i = 0u; i < pool_queues_.size(); i++) {
pool_threads_.emplace_back([this]() {
const auto thread_index = thread_index_++;
auto &pool_queue = pool_queues_[thread_index];
auto &queue = pool_queue->queue;
auto &queue_mutex = pool_queue->mutex;
auto &queue_notify = pool_queue->notify;
unique_mutex_lock queue_lock(queue_mutex);
queue_notify.notify_all();
queue_lock.unlock();
while (not shutdown_) {
queue_lock.lock();
if (queue.empty()) {
queue_notify.wait(queue_lock);
}
while (not queue.empty()) {
auto item = queue.front();
queue.pop_front();
queue_notify.notify_all();
queue_lock.unlock();
try {
const auto result = item->work();
item->work_complete(result);
} catch (const std::exception &e) {
item->work_complete(utils::from_api_error(api_error::error));
utils::error::raise_error(__FUNCTION__, e,
"exception occurred in work item");
}
queue_lock.lock();
}
queue_notify.notify_all();
queue_lock.unlock();
}
queue_lock.lock();
while (not queue.empty()) {
auto wi = queue.front();
queue.pop_front();
queue_notify.notify_all();
queue_lock.unlock();
wi->work_complete(utils::from_api_error(api_error::download_stopped));
queue_lock.lock();
}
queue_notify.notify_all();
queue_lock.unlock();
});
}
}
void client_pool::pool::shutdown() {
shutdown_ = true;
for (auto &pool_queue : pool_queues_) {
unique_mutex_lock l(pool_queue->mutex);
pool_queue->notify.notify_all();
}
for (auto &thread : pool_threads_) {
thread.join();
}
pool_queues_.clear();
pool_threads_.clear();
}
void client_pool::execute(const std::string &client_id, std::uint64_t thread_id,
const worker_callback &worker,
const worker_complete_callback &worker_complete) {
unique_mutex_lock pool_lock(pool_mutex_);
if (shutdown_) {
pool_lock.unlock();
throw std::runtime_error("Client pool is shutdown");
}
if (not pool_lookup_[client_id]) {
pool_lookup_[client_id] = std::make_shared<pool>(pool_size_);
}
pool_lookup_[client_id]->execute(thread_id, worker, worker_complete);
pool_lock.unlock();
}
void client_pool::remove_client(const std::string &client_id) {
mutex_lock pool_lock(pool_mutex_);
pool_lookup_.erase(client_id);
}
void client_pool::shutdown() {
if (not shutdown_) {
event_system::instance().raise<service_shutdown_begin>("client_pool");
unique_mutex_lock pool_lock(pool_mutex_);
if (not shutdown_) {
shutdown_ = true;
for (auto &kv : pool_lookup_) {
kv.second->shutdown();
}
pool_lookup_.clear();
}
pool_lock.unlock();
event_system::instance().raise<service_shutdown_end>("client_pool");
}
}
} // namespace repertory
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/packet/client_pool.hpp"
#include "events/event_system.hpp"
#include "events/events.hpp"
#include "utils/error_utils.hpp"
namespace repertory {
void client_pool::pool::execute(
std::uint64_t thread_id, const worker_callback &worker,
const worker_complete_callback &worker_complete) {
const auto index = thread_id % pool_queues_.size();
auto job = std::make_shared<work_item>(worker, worker_complete);
auto &pool_queue = pool_queues_[index];
unique_mutex_lock queue_lock(pool_queue->mutex);
pool_queue->queue.emplace_back(job);
pool_queue->notify.notify_all();
queue_lock.unlock();
}
client_pool::pool::pool(std::uint8_t pool_size) {
event_system::instance().raise<service_started>("client_pool");
for (std::uint8_t i = 0U; i < pool_size; i++) {
pool_queues_.emplace_back(std::make_unique<work_queue>());
}
for (std::size_t i = 0U; i < pool_queues_.size(); i++) {
pool_threads_.emplace_back([this]() {
const auto thread_index = thread_index_++;
auto &pool_queue = pool_queues_[thread_index];
auto &queue = pool_queue->queue;
auto &queue_mutex = pool_queue->mutex;
auto &queue_notify = pool_queue->notify;
unique_mutex_lock queue_lock(queue_mutex);
queue_notify.notify_all();
queue_lock.unlock();
while (not shutdown_) {
queue_lock.lock();
if (queue.empty()) {
queue_notify.wait(queue_lock);
}
while (not queue.empty()) {
auto item = queue.front();
queue.pop_front();
queue_notify.notify_all();
queue_lock.unlock();
try {
const auto result = item->work();
item->work_complete(result);
} catch (const std::exception &e) {
item->work_complete(utils::from_api_error(api_error::error));
utils::error::raise_error(__FUNCTION__, e,
"exception occurred in work item");
}
queue_lock.lock();
}
queue_notify.notify_all();
queue_lock.unlock();
}
queue_lock.lock();
while (not queue.empty()) {
auto job = queue.front();
queue.pop_front();
queue_notify.notify_all();
queue_lock.unlock();
job->work_complete(utils::from_api_error(api_error::download_stopped));
queue_lock.lock();
}
queue_notify.notify_all();
queue_lock.unlock();
});
}
}
void client_pool::pool::shutdown() {
shutdown_ = true;
for (auto &pool_queue : pool_queues_) {
mutex_lock lock(pool_queue->mutex);
pool_queue->notify.notify_all();
}
for (auto &thread : pool_threads_) {
thread.join();
}
pool_queues_.clear();
pool_threads_.clear();
}
void client_pool::execute(const std::string &client_id, std::uint64_t thread_id,
const worker_callback &worker,
const worker_complete_callback &worker_complete) {
unique_mutex_lock pool_lock(pool_mutex_);
if (shutdown_) {
pool_lock.unlock();
throw std::runtime_error("Client pool is shutdown");
}
if (not pool_lookup_[client_id]) {
pool_lookup_[client_id] = std::make_shared<pool>(pool_size_);
}
pool_lookup_[client_id]->execute(thread_id, worker, worker_complete);
pool_lock.unlock();
}
void client_pool::remove_client(const std::string &client_id) {
mutex_lock pool_lock(pool_mutex_);
pool_lookup_.erase(client_id);
}
void client_pool::shutdown() {
if (not shutdown_) {
event_system::instance().raise<service_shutdown_begin>("client_pool");
unique_mutex_lock pool_lock(pool_mutex_);
if (not shutdown_) {
shutdown_ = true;
for (auto &pool_entry : pool_lookup_) {
pool_entry.second->shutdown();
}
pool_lookup_.clear();
}
pool_lock.unlock();
event_system::instance().raise<service_shutdown_end>("client_pool");
}
}
} // namespace repertory

1138
src/comm/packet/packet.cpp
View File

File diff suppressed because it is too large Load Diff

494
src/comm/packet/packet_client.cpp
View File

@@ -1,247 +1,247 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/packet/packet_client.hpp"
#include "events/events.hpp"
#include "types/repertory.hpp"
#include "utils/error_utils.hpp"
#include "utils/timeout.hpp"
namespace repertory {
// clang-format off
E_SIMPLE2(packet_client_timeout, error, true,
std::string, event_name, en, E_STRING,
std::string, message, msg, E_STRING
);
// clang-format on
packet_client::packet_client(std::string host_name_or_ip,
std::uint8_t max_connections, std::uint16_t port,
std::uint16_t receive_timeout,
std::uint16_t send_timeout,
std::string encryption_token)
: io_context_(),
host_name_or_ip_(std::move(host_name_or_ip)),
max_connections_(max_connections ? max_connections : 20u),
port_(port),
receive_timeout_(receive_timeout),
send_timeout_(send_timeout),
encryption_token_(std::move(encryption_token)),
unique_id_(utils::create_uuid_string()) {}
packet_client::~packet_client() {
allow_connections_ = false;
close_all();
io_context_.stop();
}
void packet_client::close(client &cli) const {
try {
boost::system::error_code ec;
cli.socket.close(ec);
} catch (...) {
}
}
void packet_client::close_all() {
unique_mutex_lock clients_lock(clients_mutex_);
for (auto &c : clients_) {
close(*c.get());
}
clients_.clear();
unique_id_ = utils::create_uuid_string();
}
void packet_client::connect(client &c) {
try {
resolve();
boost::asio::connect(c.socket, resolve_results_);
c.socket.set_option(boost::asio::ip::tcp::no_delay(true));
c.socket.set_option(boost::asio::socket_base::linger(false, 0));
packet response;
const auto res = read_packet(c, response);
if (res != 0) {
throw std::runtime_error(std::to_string(res));
}
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "connection handshake failed");
}
}
auto packet_client::get_client() -> std::shared_ptr<packet_client::client> {
std::shared_ptr<client> ret;
unique_mutex_lock clients_lock(clients_mutex_);
if (allow_connections_) {
if (clients_.empty()) {
clients_lock.unlock();
ret = std::make_shared<client>(io_context_);
connect(*ret);
} else {
ret = clients_[0u];
utils::remove_element_from(clients_, ret);
clients_lock.unlock();
}
}
return ret;
}
void packet_client::put_client(std::shared_ptr<client> &c) {
mutex_lock clientsLock(clients_mutex_);
if (clients_.size() < max_connections_) {
clients_.emplace_back(c);
}
}
auto packet_client::read_packet(client &c, packet &response)
-> packet::error_type {
data_buffer buffer(sizeof(std::uint32_t));
const auto read_buffer = [&]() {
std::uint32_t offset = 0u;
while (offset < buffer.size()) {
const auto bytes_read = boost::asio::read(
c.socket,
boost::asio::buffer(&buffer[offset], buffer.size() - offset));
if (bytes_read <= 0) {
throw std::runtime_error("read failed|" + std::to_string(bytes_read));
}
offset += static_cast<std::uint32_t>(bytes_read);
}
};
read_buffer();
const auto size = boost::endian::big_to_native(
*reinterpret_cast<std::uint32_t *>(&buffer[0u]));
buffer.resize(size);
read_buffer();
response = std::move(buffer);
auto ret = response.decrypt(encryption_token_);
if (ret == 0) {
ret = response.decode(c.nonce);
}
return ret;
}
void packet_client::resolve() {
if (resolve_results_.empty()) {
resolve_results_ = tcp::resolver(io_context_)
.resolve({host_name_or_ip_, std::to_string(port_)});
}
}
auto packet_client::send(const std::string &method,
std::uint32_t &service_flags) -> packet::error_type {
packet request;
return send(method, request, service_flags);
}
auto packet_client::send(const std::string &method, packet &request,
std::uint32_t &service_flags) -> packet::error_type {
packet response;
return send(method, request, response, service_flags);
}
auto packet_client::send(const std::string &method, packet &request,
packet &response, std::uint32_t &service_flags)
-> packet::error_type {
auto success = false;
packet::error_type ret = utils::from_api_error(api_error::error);
request.encode_top(method);
request.encode_top(utils::get_thread_id());
request.encode_top(unique_id_);
request.encode_top(PACKET_SERVICE_FLAGS);
request.encode_top(get_repertory_version());
static const std::uint8_t max_attempts = 5u;
for (std::uint8_t i = 1u;
allow_connections_ && not success && (i <= max_attempts); i++) {
auto c = get_client();
if (c) {
try {
request.encode_top(c->nonce);
request.encrypt(encryption_token_);
timeout request_timeout(
[this, method, c]() {
event_system::instance().raise<packet_client_timeout>("request",
method);
close(*c.get());
},
std::chrono::seconds(send_timeout_));
std::uint32_t offset = 0u;
while (offset < request.get_size()) {
const auto bytes_written = boost::asio::write(
c->socket, boost::asio::buffer(&request[offset],
request.get_size() - offset));
if (bytes_written <= 0) {
throw std::runtime_error("write failed|" +
std::to_string(bytes_written));
}
offset += static_cast<std::uint32_t>(bytes_written);
}
request_timeout.disable();
timeout response_timeout(
[this, method, c]() {
event_system::instance().raise<packet_client_timeout>("response",
method);
close(*c.get());
},
std::chrono::seconds(receive_timeout_));
ret = read_packet(*c, response);
response_timeout.disable();
if (ret == 0) {
if ((ret = response.decode(service_flags)) == 0) {
packet::error_type res{};
if ((ret = response.decode(res)) == 0) {
ret = res;
success = true;
put_client(c);
}
}
}
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "send failed");
close_all();
if (allow_connections_ && (i < max_attempts)) {
std::this_thread::sleep_for(1s);
}
}
}
if (not allow_connections_) {
ret = utils::from_api_error(api_error::error);
success = true;
}
}
return CONVERT_STATUS_NOT_IMPLEMENTED(ret);
}
} // namespace repertory
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/packet/packet_client.hpp"
#include "events/events.hpp"
#include "types/repertory.hpp"
#include "utils/error_utils.hpp"
#include "utils/timeout.hpp"
namespace repertory {
// clang-format off
E_SIMPLE2(packet_client_timeout, error, true,
std::string, event_name, en, E_STRING,
std::string, message, msg, E_STRING
);
// clang-format on
packet_client::packet_client(std::string host_name_or_ip,
std::uint8_t max_connections, std::uint16_t port,
std::uint16_t receive_timeout,
std::uint16_t send_timeout,
std::string encryption_token)
: io_context_(),
host_name_or_ip_(std::move(host_name_or_ip)),
max_connections_(max_connections ? max_connections : 20u),
port_(port),
receive_timeout_(receive_timeout),
send_timeout_(send_timeout),
encryption_token_(std::move(encryption_token)),
unique_id_(utils::create_uuid_string()) {}
packet_client::~packet_client() {
allow_connections_ = false;
close_all();
io_context_.stop();
}
void packet_client::close(client &cli) const {
try {
boost::system::error_code ec;
cli.socket.close(ec);
} catch (...) {
}
}
void packet_client::close_all() {
unique_mutex_lock clients_lock(clients_mutex_);
for (auto &c : clients_) {
close(*c.get());
}
clients_.clear();
unique_id_ = utils::create_uuid_string();
}
void packet_client::connect(client &c) {
try {
resolve();
boost::asio::connect(c.socket, resolve_results_);
c.socket.set_option(boost::asio::ip::tcp::no_delay(true));
c.socket.set_option(boost::asio::socket_base::linger(false, 0));
packet response;
const auto res = read_packet(c, response);
if (res != 0) {
throw std::runtime_error(std::to_string(res));
}
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "connection handshake failed");
}
}
auto packet_client::get_client() -> std::shared_ptr<packet_client::client> {
std::shared_ptr<client> ret;
unique_mutex_lock clients_lock(clients_mutex_);
if (allow_connections_) {
if (clients_.empty()) {
clients_lock.unlock();
ret = std::make_shared<client>(io_context_);
connect(*ret);
} else {
ret = clients_[0u];
utils::remove_element_from(clients_, ret);
clients_lock.unlock();
}
}
return ret;
}
void packet_client::put_client(std::shared_ptr<client> &c) {
mutex_lock clientsLock(clients_mutex_);
if (clients_.size() < max_connections_) {
clients_.emplace_back(c);
}
}
auto packet_client::read_packet(client &c, packet &response)
-> packet::error_type {
data_buffer buffer(sizeof(std::uint32_t));
const auto read_buffer = [&]() {
std::uint32_t offset = 0u;
while (offset < buffer.size()) {
const auto bytes_read = boost::asio::read(
c.socket,
boost::asio::buffer(&buffer[offset], buffer.size() - offset));
if (bytes_read <= 0) {
throw std::runtime_error("read failed|" + std::to_string(bytes_read));
}
offset += static_cast<std::uint32_t>(bytes_read);
}
};
read_buffer();
const auto size = boost::endian::big_to_native(
*reinterpret_cast<std::uint32_t *>(&buffer[0u]));
buffer.resize(size);
read_buffer();
response = std::move(buffer);
auto ret = response.decrypt(encryption_token_);
if (ret == 0) {
ret = response.decode(c.nonce);
}
return ret;
}
void packet_client::resolve() {
if (resolve_results_.empty()) {
resolve_results_ = tcp::resolver(io_context_)
.resolve({host_name_or_ip_, std::to_string(port_)});
}
}
auto packet_client::send(const std::string &method,
std::uint32_t &service_flags) -> packet::error_type {
packet request;
return send(method, request, service_flags);
}
auto packet_client::send(const std::string &method, packet &request,
std::uint32_t &service_flags) -> packet::error_type {
packet response;
return send(method, request, response, service_flags);
}
auto packet_client::send(const std::string &method, packet &request,
packet &response, std::uint32_t &service_flags)
-> packet::error_type {
auto success = false;
packet::error_type ret = utils::from_api_error(api_error::error);
request.encode_top(method);
request.encode_top(utils::get_thread_id());
request.encode_top(unique_id_);
request.encode_top(PACKET_SERVICE_FLAGS);
request.encode_top(get_repertory_version());
static const std::uint8_t max_attempts = 5u;
for (std::uint8_t i = 1u;
allow_connections_ && not success && (i <= max_attempts); i++) {
auto c = get_client();
if (c) {
try {
request.encode_top(c->nonce);
request.encrypt(encryption_token_);
timeout request_timeout(
[this, method, c]() {
event_system::instance().raise<packet_client_timeout>("request",
method);
close(*c.get());
},
std::chrono::seconds(send_timeout_));
std::uint32_t offset = 0u;
while (offset < request.get_size()) {
const auto bytes_written = boost::asio::write(
c->socket, boost::asio::buffer(&request[offset],
request.get_size() - offset));
if (bytes_written <= 0) {
throw std::runtime_error("write failed|" +
std::to_string(bytes_written));
}
offset += static_cast<std::uint32_t>(bytes_written);
}
request_timeout.disable();
timeout response_timeout(
[this, method, c]() {
event_system::instance().raise<packet_client_timeout>("response",
method);
close(*c.get());
},
std::chrono::seconds(receive_timeout_));
ret = read_packet(*c, response);
response_timeout.disable();
if (ret == 0) {
if ((ret = response.decode(service_flags)) == 0) {
packet::error_type res{};
if ((ret = response.decode(res)) == 0) {
ret = res;
success = true;
put_client(c);
}
}
}
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "send failed");
close_all();
if (allow_connections_ && (i < max_attempts)) {
std::this_thread::sleep_for(1s);
}
}
}
if (not allow_connections_) {
ret = utils::from_api_error(api_error::error);
success = true;
}
}
return CONVERT_STATUS_NOT_IMPLEMENTED(ret);
}
} // namespace repertory

500
src/comm/packet/packet_server.cpp
View File

@@ -1,250 +1,250 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/packet/packet_server.hpp"
#include "comm/packet/packet.hpp"
#include "events/event_system.hpp"
#include "events/events.hpp"
#include "types/repertory.hpp"
#include "utils/error_utils.hpp"
#include "utils/string_utils.hpp"
#include "utils/utils.hpp"
namespace repertory {
using std::thread;
packet_server::packet_server(std::uint16_t port, std::string token,
std::uint8_t pool_size, closed_callback closed,
message_handler_callback message_handler)
: encryption_token_(std::move(token)),
closed_(std::move(closed)),
message_handler_(std::move(message_handler)) {
initialize(port, pool_size);
event_system::instance().raise<service_started>("packet_server");
}
packet_server::~packet_server() {
event_system::instance().raise<service_shutdown_begin>("packet_server");
std::thread([this]() {
for (std::size_t i = 0u; i < service_threads_.size(); i++) {
io_context_.stop();
}
}).detach();
server_thread_->join();
server_thread_.reset();
event_system::instance().raise<service_shutdown_end>("packet_server");
}
void packet_server::add_client(connection &c, const std::string &client_id) {
c.client_id = client_id;
recur_mutex_lock connection_lock(connection_mutex_);
if (connection_lookup_.find(client_id) == connection_lookup_.end()) {
connection_lookup_[client_id] = 1u;
} else {
connection_lookup_[client_id]++;
}
}
void packet_server::initialize(const uint16_t &port, uint8_t pool_size) {
pool_size = std::max(uint8_t(1u), pool_size);
server_thread_ = std::make_unique<std::thread>([this, port, pool_size]() {
tcp::acceptor acceptor(io_context_);
try {
const auto endpoint = tcp::endpoint(tcp::v4(), port);
acceptor.open(endpoint.protocol());
acceptor.set_option(socket_base::reuse_address(true));
acceptor.bind(endpoint);
acceptor.listen();
} catch (const std::exception &e) {
repertory::utils::error::raise_error(__FUNCTION__, e,
"exception occurred");
}
listen_for_connection(acceptor);
for (std::uint8_t i = 0u; i < pool_size; i++) {
service_threads_.emplace_back([this]() { io_context_.run(); });
}
for (auto &th : service_threads_) {
th.join();
}
});
}
void packet_server::listen_for_connection(tcp::acceptor &acceptor) {
auto c = std::make_shared<packet_server::connection>(io_context_, acceptor);
acceptor.async_accept(c->socket,
boost::bind(&packet_server::on_accept, this, c,
boost::asio::placeholders::error));
}
void packet_server::on_accept(std::shared_ptr<connection> c,
boost::system::error_code ec) {
listen_for_connection(c->acceptor);
if (ec) {
utils::error::raise_error(__FUNCTION__, ec.message());
std::this_thread::sleep_for(1s);
} else {
c->socket.set_option(boost::asio::ip::tcp::no_delay(true));
c->socket.set_option(boost::asio::socket_base::linger(false, 0));
c->generate_nonce();
packet response;
send_response(c, 0, response);
}
}
void packet_server::read_header(std::shared_ptr<connection> c) {
static const std::string function_name = __FUNCTION__;
c->buffer.resize(sizeof(std::uint32_t));
boost::asio::async_read(
c->socket, boost::asio::buffer(&c->buffer[0u], c->buffer.size()),
[this, c](boost::system::error_code ec, std::size_t) {
if (ec) {
remove_client(*c);
repertory::utils::error::raise_error(function_name, ec.message());
} else {
auto to_read = *reinterpret_cast<std::uint32_t *>(&c->buffer[0u]);
boost::endian::big_to_native_inplace(to_read);
read_packet(c, to_read);
}
});
}
void packet_server::read_packet(std::shared_ptr<connection> c,
std::uint32_t data_size) {
try {
const auto read_buffer = [&]() {
std::uint32_t offset = 0u;
while (offset < c->buffer.size()) {
const auto bytes_read = boost::asio::read(
c->socket,
boost::asio::buffer(&c->buffer[offset], c->buffer.size() - offset));
if (bytes_read <= 0) {
throw std::runtime_error("read failed|" + std::to_string(bytes_read));
}
offset += static_cast<std::uint32_t>(bytes_read);
}
};
auto should_send_response = true;
auto response = std::make_shared<packet>();
c->buffer.resize(data_size);
read_buffer();
packet::error_type ret;
auto request = std::make_shared<packet>(c->buffer);
if (request->decrypt(encryption_token_) == 0) {
std::string nonce;
if ((ret = request->decode(nonce)) == 0) {
if (nonce != c->nonce) {
throw std::runtime_error("invalid nonce");
}
c->generate_nonce();
std::string version;
if ((ret = request->decode(version)) == 0) {
if (utils::compare_version_strings(
version, REPERTORY_MIN_REMOTE_VERSION) >= 0) {
std::uint32_t service_flags = 0u;
DECODE_OR_IGNORE(request, service_flags);
std::string client_id;
DECODE_OR_IGNORE(request, client_id);
std::uint64_t thread_id = 0u;
DECODE_OR_IGNORE(request, thread_id);
std::string method;
DECODE_OR_IGNORE(request, method);
if (ret == 0) {
if (c->client_id.empty()) {
add_client(*c, client_id);
}
should_send_response = false;
message_handler_(service_flags, client_id, thread_id, method,
request.get(), *response,
[this, c, request,
response](const packet::error_type &result) {
this->send_response(c, result, *response);
});
}
} else {
ret = utils::from_api_error(api_error::incompatible_version);
}
} else {
ret = utils::from_api_error(api_error::invalid_version);
}
} else {
throw std::runtime_error("invalid nonce");
}
} else {
throw std::runtime_error("decryption failed");
}
if (should_send_response) {
send_response(c, ret, *response);
}
} catch (const std::exception &e) {
remove_client(*c);
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
}
void packet_server::remove_client(connection &c) {
if (not c.client_id.empty()) {
recur_mutex_lock connection_lock(connection_mutex_);
if (not --connection_lookup_[c.client_id]) {
connection_lookup_.erase(c.client_id);
closed_(c.client_id);
}
}
}
void packet_server::send_response(std::shared_ptr<connection> c,
const packet::error_type &result,
packet &response) {
static const std::string function_name = __FUNCTION__;
response.encode_top(result);
response.encode_top(PACKET_SERVICE_FLAGS);
response.encode_top(c->nonce);
response.encrypt(encryption_token_);
response.transfer_into(c->buffer);
boost::asio::async_write(
c->socket, boost::asio::buffer(c->buffer),
[this, c](boost::system::error_code ec, std::size_t /*length*/) {
if (ec) {
remove_client(*c);
utils::error::raise_error(function_name, ec.message());
} else {
read_header(c);
}
});
}
} // namespace repertory
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include "comm/packet/packet_server.hpp"
#include "comm/packet/packet.hpp"
#include "events/event_system.hpp"
#include "events/events.hpp"
#include "types/repertory.hpp"
#include "utils/error_utils.hpp"
#include "utils/string_utils.hpp"
#include "utils/utils.hpp"
namespace repertory {
using std::thread;
packet_server::packet_server(std::uint16_t port, std::string token,
std::uint8_t pool_size, closed_callback closed,
message_handler_callback message_handler)
: encryption_token_(std::move(token)),
closed_(std::move(closed)),
message_handler_(std::move(message_handler)) {
initialize(port, pool_size);
event_system::instance().raise<service_started>("packet_server");
}
packet_server::~packet_server() {
event_system::instance().raise<service_shutdown_begin>("packet_server");
std::thread([this]() {
for (std::size_t i = 0u; i < service_threads_.size(); i++) {
io_context_.stop();
}
}).detach();
server_thread_->join();
server_thread_.reset();
event_system::instance().raise<service_shutdown_end>("packet_server");
}
void packet_server::add_client(connection &c, const std::string &client_id) {
c.client_id = client_id;
recur_mutex_lock connection_lock(connection_mutex_);
if (connection_lookup_.find(client_id) == connection_lookup_.end()) {
connection_lookup_[client_id] = 1u;
} else {
connection_lookup_[client_id]++;
}
}
void packet_server::initialize(const uint16_t &port, uint8_t pool_size) {
pool_size = std::max(uint8_t(1u), pool_size);
server_thread_ = std::make_unique<std::thread>([this, port, pool_size]() {
tcp::acceptor acceptor(io_context_);
try {
const auto endpoint = tcp::endpoint(tcp::v4(), port);
acceptor.open(endpoint.protocol());
acceptor.set_option(socket_base::reuse_address(true));
acceptor.bind(endpoint);
acceptor.listen();
} catch (const std::exception &e) {
repertory::utils::error::raise_error(__FUNCTION__, e,
"exception occurred");
}
listen_for_connection(acceptor);
for (std::uint8_t i = 0u; i < pool_size; i++) {
service_threads_.emplace_back([this]() { io_context_.run(); });
}
for (auto &th : service_threads_) {
th.join();
}
});
}
void packet_server::listen_for_connection(tcp::acceptor &acceptor) {
auto c = std::make_shared<packet_server::connection>(io_context_, acceptor);
acceptor.async_accept(c->socket,
boost::bind(&packet_server::on_accept, this, c,
boost::asio::placeholders::error));
}
void packet_server::on_accept(std::shared_ptr<connection> c,
boost::system::error_code ec) {
listen_for_connection(c->acceptor);
if (ec) {
utils::error::raise_error(__FUNCTION__, ec.message());
std::this_thread::sleep_for(1s);
} else {
c->socket.set_option(boost::asio::ip::tcp::no_delay(true));
c->socket.set_option(boost::asio::socket_base::linger(false, 0));
c->generate_nonce();
packet response;
send_response(c, 0, response);
}
}
void packet_server::read_header(std::shared_ptr<connection> c) {
static const std::string function_name = __FUNCTION__;
c->buffer.resize(sizeof(std::uint32_t));
boost::asio::async_read(
c->socket, boost::asio::buffer(&c->buffer[0u], c->buffer.size()),
[this, c](boost::system::error_code ec, std::size_t) {
if (ec) {
remove_client(*c);
repertory::utils::error::raise_error(function_name, ec.message());
} else {
auto to_read = *reinterpret_cast<std::uint32_t *>(&c->buffer[0u]);
boost::endian::big_to_native_inplace(to_read);
read_packet(c, to_read);
}
});
}
void packet_server::read_packet(std::shared_ptr<connection> c,
std::uint32_t data_size) {
try {
const auto read_buffer = [&]() {
std::uint32_t offset = 0u;
while (offset < c->buffer.size()) {
const auto bytes_read = boost::asio::read(
c->socket,
boost::asio::buffer(&c->buffer[offset], c->buffer.size() - offset));
if (bytes_read <= 0) {
throw std::runtime_error("read failed|" + std::to_string(bytes_read));
}
offset += static_cast<std::uint32_t>(bytes_read);
}
};
auto should_send_response = true;
auto response = std::make_shared<packet>();
c->buffer.resize(data_size);
read_buffer();
packet::error_type ret;
auto request = std::make_shared<packet>(c->buffer);
if (request->decrypt(encryption_token_) == 0) {
std::string nonce;
if ((ret = request->decode(nonce)) == 0) {
if (nonce != c->nonce) {
throw std::runtime_error("invalid nonce");
}
c->generate_nonce();
std::string version;
if ((ret = request->decode(version)) == 0) {
if (utils::compare_version_strings(
version, REPERTORY_MIN_REMOTE_VERSION) >= 0) {
std::uint32_t service_flags = 0u;
DECODE_OR_IGNORE(request, service_flags);
std::string client_id;
DECODE_OR_IGNORE(request, client_id);
std::uint64_t thread_id = 0u;
DECODE_OR_IGNORE(request, thread_id);
std::string method;
DECODE_OR_IGNORE(request, method);
if (ret == 0) {
if (c->client_id.empty()) {
add_client(*c, client_id);
}
should_send_response = false;
message_handler_(service_flags, client_id, thread_id, method,
request.get(), *response,
[this, c, request,
response](const packet::error_type &result) {
this->send_response(c, result, *response);
});
}
} else {
ret = utils::from_api_error(api_error::incompatible_version);
}
} else {
ret = utils::from_api_error(api_error::invalid_version);
}
} else {
throw std::runtime_error("invalid nonce");
}
} else {
throw std::runtime_error("decryption failed");
}
if (should_send_response) {
send_response(c, ret, *response);
}
} catch (const std::exception &e) {
remove_client(*c);
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
}
void packet_server::remove_client(connection &c) {
if (not c.client_id.empty()) {
recur_mutex_lock connection_lock(connection_mutex_);
if (not --connection_lookup_[c.client_id]) {
connection_lookup_.erase(c.client_id);
closed_(c.client_id);
}
}
}
void packet_server::send_response(std::shared_ptr<connection> c,
const packet::error_type &result,
packet &response) {
static const std::string function_name = __FUNCTION__;
response.encode_top(result);
response.encode_top(PACKET_SERVICE_FLAGS);
response.encode_top(c->nonce);
response.encrypt(encryption_token_);
response.transfer_into(c->buffer);
boost::asio::async_write(
c->socket, boost::asio::buffer(c->buffer),
[this, c](boost::system::error_code ec, std::size_t /*length*/) {
if (ec) {
remove_client(*c);
utils::error::raise_error(function_name, ec.message());
} else {
read_header(c);
}
});
}
} // namespace repertory

575
src/comm/s3/s3_comm.cpp
View File

@@ -1,575 +0,0 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#if defined(REPERTORY_ENABLE_S3)
#include "comm/s3/s3_comm.hpp"
#include "app_config.hpp"
#include "comm/curl/curl_comm.hpp"
#include "comm/s3/s3_requests.hpp"
#include "events/event_system.hpp"
#include "events/events.hpp"
#include "providers/i_provider.hpp"
#include "types/repertory.hpp"
#include "types/s3.hpp"
#include "utils/encryption.hpp"
#include "utils/error_utils.hpp"
#include "utils/path_utils.hpp"
#include "utils/polling.hpp"
#include "utils/string_utils.hpp"
namespace repertory {
static const get_key_callback empty_key = []() { return ""; };
s3_comm::s3_comm(const app_config &config)
: config_(config), s3_config_(config.get_s3_config()) {
s3_config_.bucket = utils::string::trim(s3_config_.bucket);
// TODO make configurable
const auto enable_path_style =
utils::string::begins_with(s3_config_.url, "http://localhost") ||
utils::string::begins_with(s3_config_.url, "https://localhost") ||
utils::string::begins_with(s3_config_.url, "http://127.0.0.1") ||
utils::string::begins_with(s3_config_.url, "https://127.0.0.1");
s3_client_ = std::make_unique<curl_comm>(s3_config_);
s3_client_->enable_s3_path_style(enable_path_style);
polling::instance().set_callback(
{"s3_directory_cache", polling::frequency::high,
[this]() { this->clear_expired_directories(); }});
}
s3_comm::s3_comm(s3_comm &&comm)
: config_(std::move(comm.config_)),
s3_config_(std::move(comm.s3_config_)),
s3_client_(std::move(comm.s3_client_)) {
comm.active_ = false;
polling::instance().set_callback(
{"s3_directory_cache", polling::frequency::high,
[this]() { this->clear_expired_directories(); }});
}
s3_comm::~s3_comm() {
if (active_) {
polling::instance().remove_callback("s3_directory_cache");
}
}
void s3_comm::clear_expired_directories() {
recur_mutex_lock l(cached_directories_mutex_);
std::vector<std::string> expired_list;
for (const auto &kv : cached_directories_) {
if (kv.second.expiration <= std::chrono::system_clock::now()) {
expired_list.emplace_back(kv.first);
}
}
for (const auto &expired : expired_list) {
event_system::instance().raise<debug_log>(__FUNCTION__, expired, "expired");
cached_directories_.erase(expired);
}
}
auto s3_comm::create_directory(const std::string &api_path) -> api_error {
raise_begin(__FUNCTION__, api_path);
long response_code{};
auto object_name = get_object_name(api_path, empty_key) + '/';
if (not create_directory_object_request(*s3_client_, s3_config_, object_name,
response_code)) {
return raise_end(__FUNCTION__, api_path, api_error::comm_error,
response_code);
}
return raise_end(__FUNCTION__, api_path,
response_code == 200 ? api_error::success
: api_error::comm_error,
response_code);
}
auto s3_comm::directory_exists(const std::string &api_path) const -> api_error {
raise_begin(__FUNCTION__, api_path);
auto object_name = get_object_name(api_path, empty_key) + "/";
head_object_result result{};
long response_code{};
if (head_object_request(*s3_client_, s3_config_, object_name, result,
response_code)) {
if (response_code == 404) {
return raise_end(__FUNCTION__, api_path, api_error::directory_not_found,
response_code);
}
return raise_end(__FUNCTION__, api_path, api_error::directory_exists,
response_code);
}
return raise_end(__FUNCTION__, api_path, api_error::comm_error,
response_code);
}
auto s3_comm::file_exists(const std::string &api_path,
const get_key_callback &get_key) const -> api_error {
raise_begin(__FUNCTION__, api_path);
if (get_cached_file_exists(api_path)) {
return raise_end(__FUNCTION__, api_path, api_error::item_exists, 200);
}
auto object_name = get_object_name(api_path, get_key);
head_object_result result{};
long response_code{};
if (head_object_request(*s3_client_, s3_config_, object_name, result,
response_code)) {
if (response_code == 404) {
return raise_end(__FUNCTION__, api_path, api_error::item_not_found,
response_code);
}
return raise_end(__FUNCTION__, api_path, api_error::item_exists,
response_code);
}
return raise_end(__FUNCTION__, api_path, api_error::directory_exists,
response_code);
}
auto s3_comm::get_object_list(std::vector<directory_item> &list) const
-> api_error {
raise_begin(__FUNCTION__, "/");
long response_code{};
auto success =
list_objects_request(*s3_client_, s3_config_, list, response_code);
return raise_end(__FUNCTION__, "/",
success ? api_error::success : api_error::comm_error,
response_code);
}
auto s3_comm::get_object_name(const std::string &api_path,
const get_key_callback &get_key) const
-> std::string {
auto object_name = utils::path::create_api_path(api_path).substr(1);
const auto key = get_key();
if (not key.empty()) {
auto parts = utils::string::split(object_name, '/', false);
parts[parts.size() - 1u] = key;
object_name = utils::string::join(parts, '/');
}
return object_name;
}
auto s3_comm::get_cached_directory_item_count(const std::string &api_path,
std::size_t &count) const
-> bool {
recur_mutex_lock l(cached_directories_mutex_);
if (cached_directories_.find(api_path) != cached_directories_.end()) {
count = cached_directories_.at(api_path).items.size();
return true;
}
return false;
}
auto s3_comm::get_cached_directory_items(const std::string &api_path,
meta_provider_callback meta_provider,
directory_item_list &list) const
-> bool {
unique_recur_mutex_lock l(cached_directories_mutex_);
if (cached_directories_.find(api_path) != cached_directories_.end()) {
auto &cachedEntry = cached_directories_.at(api_path);
list = cachedEntry.items;
cached_directories_[api_path].reset_timeout(
std::chrono::seconds(config_.get_s3_config().cache_timeout_secs));
l.unlock();
for (auto &item : list) {
meta_provider(item);
}
return true;
}
return false;
}
auto s3_comm::get_cached_file_exists(const std::string &api_path) const
-> bool {
unique_recur_mutex_lock l(cached_directories_mutex_);
const auto parent_api_path = utils::path::get_parent_api_path(api_path);
if (cached_directories_.find(parent_api_path) != cached_directories_.end()) {
auto &entry = cached_directories_.at(parent_api_path);
if (std::find_if(entry.items.begin(), entry.items.end(),
[&api_path](const auto &item) -> bool {
return not item.directory && (api_path == item.api_path);
}) != entry.items.end()) {
cached_directories_[api_path].reset_timeout(
std::chrono::seconds(config_.get_s3_config().cache_timeout_secs));
return true;
}
}
return false;
}
auto s3_comm::get_directory_item_count(
const std::string &api_path, meta_provider_callback meta_provider) const
-> std::size_t {
raise_begin(__FUNCTION__, api_path);
std::size_t ret = 0u;
if (not get_cached_directory_item_count(api_path, ret)) {
directory_item_list list;
const auto res = grab_directory_items(api_path, meta_provider, list);
if (res != api_error::success) {
utils::error::raise_api_path_error(__FUNCTION__, api_path, res,
"failed to grab directory items");
}
return list.size();
}
if (config_.get_event_level() >= event_level::debug) {
event_system::instance().raise<debug_log>(__FUNCTION__, api_path,
"end|" + std::to_string(ret));
}
return ret;
}
auto s3_comm::get_directory_items(const std::string &api_path,
meta_provider_callback meta_provider,
directory_item_list &list) const
-> api_error {
raise_begin(__FUNCTION__, api_path);
auto ret = api_error::success;
if (not get_cached_directory_items(api_path, meta_provider, list)) {
ret = grab_directory_items(api_path, meta_provider, list);
}
if (config_.get_event_level() >= event_level::debug) {
event_system::instance().raise<debug_log>(
__FUNCTION__, api_path, "end|" + api_error_to_string(ret));
}
return ret;
}
auto s3_comm::get_directory_list(api_file_list &list) const -> api_error {
raise_begin(__FUNCTION__, "/");
long response_code{};
auto success =
list_directories_request(*s3_client_, s3_config_, list, response_code);
return raise_end(__FUNCTION__, "/",
success ? api_error::success : api_error::comm_error,
response_code);
}
auto s3_comm::get_file(const std::string &api_path,
const get_key_callback &get_key,
const get_name_callback &get_name,
const get_token_callback &get_token,
api_file &file) const -> api_error {
raise_begin(__FUNCTION__, api_path);
auto ret = api_error::success;
auto object_name = get_object_name(api_path, get_key);
head_object_result result{};
long response_code{};
if (head_object_request(*s3_client_, s3_config_, object_name, result,
response_code)) {
const auto key = get_key();
object_name = get_name(key, object_name);
file.accessed_date = utils::get_file_time_now();
file.api_path = utils::path::create_api_path(object_name);
file.api_parent = utils::path::get_parent_api_path(file.api_path);
file.changed_date = utils::aws::format_time(result.last_modified);
file.creation_date = utils::aws::format_time(result.last_modified);
file.encryption_token = get_token();
file.file_size =
file.encryption_token.empty()
? result.content_length
: utils::encryption::encrypting_reader::calculate_decrypted_size(
result.content_length);
file.modified_date = utils::aws::format_time(result.last_modified);
} else {
utils::error::raise_api_path_error(__FUNCTION__, api_path, response_code,
"head object request failed");
ret = api_error::comm_error;
}
if (config_.get_event_level() >= event_level::debug) {
event_system::instance().raise<debug_log>(
__FUNCTION__, api_path, "end|" + api_error_to_string(ret));
}
return ret;
}
auto s3_comm::get_file_list(
const get_api_file_token_callback &get_api_file_token,
const get_name_callback &get_name, api_file_list &list) const -> api_error {
raise_begin(__FUNCTION__, "/");
long response_code{};
auto success = list_files_request(*s3_client_, s3_config_, get_api_file_token,
get_name, list, response_code);
return raise_end(__FUNCTION__, "/",
success ? api_error::success : api_error::comm_error,
response_code);
}
auto s3_comm::grab_directory_items(const std::string &api_path,
meta_provider_callback meta_provider,
directory_item_list &list) const
-> api_error {
auto object_name = get_object_name(api_path, empty_key);
long response_code{};
if (list_objects_in_directory_request(*s3_client_, s3_config_, object_name,
meta_provider, list, response_code)) {
if (response_code == 404) {
return api_error::directory_not_found;
}
if (response_code != 200) {
return api_error::comm_error;
}
set_cached_directory_items(api_path, list);
return api_error::success;
}
return api_error::comm_error;
}
void s3_comm::raise_begin(const std::string &function_name,
const std::string &api_path) const {
if (config_.get_event_level() >= event_level::debug) {
event_system::instance().raise<debug_log>(function_name, api_path,
"begin|");
}
}
auto s3_comm::raise_end(const std::string &function_name,
const std::string &api_path, const api_error &error,
long code) const -> api_error {
if (config_.get_event_level() >= event_level::debug) {
event_system::instance().raise<debug_log>(
function_name, api_path,
"end|" + api_error_to_string(error) + '|' + std::to_string(code));
}
return error;
}
auto s3_comm::read_file_bytes(const std::string &api_path, std::size_t size,
std::uint64_t offset, data_buffer &data,
const get_key_callback &get_key,
const get_size_callback &get_size,
const get_token_callback &get_token,
stop_type &stop_requested) const -> api_error {
data.clear();
auto object_name = get_object_name(api_path, get_key);
const auto encryption_token = get_token();
const auto data_size = get_size();
if (encryption_token.empty()) {
long response_code{};
if (not read_object_request(*s3_client_, s3_config_, object_name, size,
offset, data, response_code, stop_requested)) {
auto res =
stop_requested ? api_error::download_stopped : api_error::comm_error;
utils::error::raise_api_path_error(__FUNCTION__, api_path, res,
"failed to read file bytes");
return res;
}
if (response_code < 200 || response_code >= 300) {
utils::error::raise_api_path_error(__FUNCTION__, api_path, response_code,
"failed to read file bytes");
return api_error::comm_error;
}
return api_error::success;
}
const auto key = utils::encryption::generate_key(encryption_token);
return utils::encryption::read_encrypted_range(
{offset, offset + size - 1}, key,
[&](data_buffer &ct, std::uint64_t start_offset,
std::uint64_t end_offset) -> api_error {
return read_file_bytes(
api_path, (end_offset - start_offset + 1u), start_offset, ct,
get_key, get_size, []() -> std::string { return ""; },
stop_requested);
},
data_size, data);
}
void s3_comm::remove_cached_directory(const std::string &api_path) {
recur_mutex_lock l(cached_directories_mutex_);
cached_directories_.erase(api_path);
}
auto s3_comm::remove_directory(const std::string &api_path) -> api_error {
raise_begin(__FUNCTION__, api_path);
auto object_name = get_object_name(api_path, empty_key) + "/";
long response_code{};
if (delete_object_request(*s3_client_, s3_config_, object_name,
response_code)) {
if (response_code == 404) {
return raise_end(__FUNCTION__, api_path, api_error::directory_not_found,
response_code);
}
if (response_code != 204) {
return raise_end(__FUNCTION__, api_path, api_error::comm_error,
response_code);
}
remove_cached_directory(utils::path::get_parent_api_path(api_path));
remove_cached_directory(api_path);
return raise_end(__FUNCTION__, api_path, api_error::success, response_code);
}
return raise_end(__FUNCTION__, api_path, api_error::comm_error,
response_code);
}
auto s3_comm::remove_file(const std::string &api_path,
const get_key_callback &get_key) -> api_error {
raise_begin(__FUNCTION__, api_path);
auto object_name = get_object_name(api_path, get_key);
long response_code{};
if (delete_object_request(*s3_client_, s3_config_, object_name,
response_code)) {
if (response_code == 404) {
return raise_end(__FUNCTION__, api_path, api_error::item_not_found,
response_code);
}
if (response_code != 204) {
return raise_end(__FUNCTION__, api_path, api_error::comm_error,
response_code);
}
remove_cached_directory(utils::path::get_parent_api_path(api_path));
return raise_end(__FUNCTION__, api_path, api_error::success, response_code);
}
return raise_end(__FUNCTION__, api_path, api_error::comm_error,
response_code);
}
auto s3_comm::rename_file(const std::string & /*api_path*/,
const std::string & /*new_api_path*/) -> api_error {
return api_error::not_implemented;
/* if (config_.get_event_level() >= event_level::debug) { */
/* event_system::instance().raise<debug_log>(__FUNCTION__, api_path,
* "begin"); */
/* } */
/* auto ret = api_error::success; */
/* */
/* std::string bucket_name, object_name; */
/* get_object_name(api_path, bucket_name, object_name); */
/* */
/* std::string new_object_name; */
/* get_object_name(new_api_path, bucket_name, new_object_name); */
/* */
/* Aws::S3::Model::CopyObjectRequest request{}; */
/* request.SetBucket(bucket_name); */
/* request.SetCopySource(bucket_name + '/' + object_name); */
/* request.SetKey(new_object_name); */
/* */
/* const auto outcome = s3_client_->CopyObject(request); */
/* if (outcome.IsSuccess()) { */
/* ret = remove_file(api_path); */
/* } else { */
/* const auto &error = outcome.GetError(); */
/* event_system::instance().raise<repertory_exception>(__FUNCTION__,
* error.GetExceptionName()
* +
* "|" + */
/* error.GetMessage());
*/
/* ret = api_error::comm_error; */
/* } */
/* */
/* if (config_.get_event_level() >= event_level::debug) { */
/* event_system::instance().raise<debug_log>(__FUNCTION__, api_path, */
/* "end|" +
* std::to_string(std::uint8_t(ret))); */
/* } */
/* return ret; */
}
void s3_comm::set_cached_directory_items(const std::string &api_path,
directory_item_list list) const {
recur_mutex_lock l(cached_directories_mutex_);
cached_directories_[api_path].items = std::move(list);
cached_directories_[api_path].reset_timeout(
std::chrono::seconds(config_.get_s3_config().cache_timeout_secs));
}
auto s3_comm::upload_file(const std::string &api_path,
const std::string &source_path,
const std::string &encryption_token,
const get_key_callback &get_key,
const set_key_callback &set_key,
stop_type &stop_requested) -> api_error {
raise_begin(__FUNCTION__, api_path);
auto object_name = get_object_name(api_path, get_key);
long response_code{};
if (not put_object_request(*s3_client_, s3_config_, object_name, source_path,
encryption_token, get_key, set_key, response_code,
stop_requested)) {
return raise_end(__FUNCTION__, api_path,
stop_requested ? api_error::upload_stopped
: api_error::upload_failed,
response_code);
}
if (response_code != 200) {
return raise_end(__FUNCTION__, api_path, api_error::comm_error,
response_code);
}
remove_cached_directory(utils::path::get_parent_api_path(api_path));
return raise_end(__FUNCTION__, api_path, api_error::success, response_code);
}
} // namespace repertory
#endif // REPERTORY_ENABLE_S3

399
src/comm/s3/s3_requests_curl.cpp
View File

@@ -1,399 +0,0 @@
/*
Copyright <2018-2023> <scott.e.graves@protonmail.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#if defined(REPERTORY_ENABLE_S3)
#include "comm/s3/s3_requests.hpp"
#include "comm/curl/curl_comm.hpp"
#include "comm/curl/requests/http_get.hpp"
#include "utils/encryption.hpp"
#include "utils/error_utils.hpp"
#include "utils/path_utils.hpp"
namespace repertory {
namespace {
[[nodiscard]] auto
get_object_list(i_http_comm &client, const s3_config &config,
std::string &response_data, long &response_code,
std::optional<std::string> delimiter = std::nullopt,
std::optional<std::string> prefix = std::nullopt) -> bool {
curl::requests::http_get get{};
get.aws_service = "aws:amz:" + config.region + ":s3";
get.path = '/';
get.query["list-type"] = "2";
if (delimiter.has_value() && not delimiter.value().empty()) {
get.query["delimiter"] = delimiter.value();
}
if (prefix.has_value() && not prefix.value().empty()) {
get.query["prefix"] = prefix.value();
}
get.response_handler = [&response_data](const data_buffer &data,
long /*response_code*/) {
response_data = std::string(data.begin(), data.end());
};
stop_type stop_requested{};
return client.make_request(get, response_code, stop_requested);
}
} // namespace
auto create_directory_object_request_impl(i_http_comm &client,
const s3_config &config,
const std::string &object_name,
long &response_code) -> bool {
try {
curl::requests::http_put_file put_file{};
put_file.aws_service = "aws:amz:" + config.region + ":s3";
put_file.file_name =
*(utils::string::split(object_name, '/', false).end() - 1U);
put_file.path = '/' + object_name;
stop_type stop_requested{false};
return client.make_request(put_file, response_code, stop_requested);
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto delete_object_request_impl(i_http_comm &client, const s3_config &config,
const std::string &object_name,
long &response_code) -> bool {
try {
head_object_result result{};
if (not head_object_request_impl(client, config, object_name, result,
response_code)) {
return false;
}
if (response_code == 404) {
return true;
}
curl::requests::http_delete del{};
del.aws_service = "aws:amz:" + config.region + ":s3";
del.path = '/' + object_name;
stop_type stop_requested{false};
return client.make_request(del, response_code, stop_requested);
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto head_object_request_impl(i_http_comm &client, const s3_config &config,
const std::string &object_name,
head_object_result &result, long &response_code)
-> bool {
try {
curl::requests::http_head head{};
head.aws_service = "aws:amz:" + config.region + ":s3";
head.path = '/' + object_name;
head.response_headers = http_headers{};
stop_type stop_requested{false};
if (not client.make_request(head, response_code, stop_requested)) {
return false;
}
if (response_code == 200) {
result.from_headers(head.response_headers.value());
}
return true;
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto list_directories_request_impl(i_http_comm &client, const s3_config &config,
list_directories_result &result,
long &response_code) -> bool {
try {
std::string response_data{};
if (not get_object_list(client, config, response_data, response_code)) {
return false;
}
if (response_code != 200) {
return false;
}
pugi::xml_document doc;
auto res = doc.load_string(response_data.c_str());
if (res.status != pugi::xml_parse_status::status_ok) {
return false;
}
auto node_list = doc.select_nodes("/ListBucketResult/Contents");
for (const auto &node : node_list) {
auto object_name =
node.node().select_node("Key").node().text().as_string();
if (utils::string::ends_with(object_name, "/")) {
api_file directory{};
directory.api_path = utils::path::create_api_path(object_name);
directory.api_parent =
utils::path::get_parent_api_path(directory.api_path);
directory.accessed_date = utils::get_file_time_now();
directory.changed_date = utils::convert_api_date(
node.node().select_node("LastModified").node().text().as_string());
directory.creation_date = directory.changed_date;
directory.modified_date = directory.changed_date;
result.emplace_back(std::move(directory));
}
}
return true;
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto list_files_request_impl(
i_http_comm &client, const s3_config &config,
const get_api_file_token_callback &get_api_file_token,
const get_name_callback &get_name, list_files_result &result,
long &response_code) -> bool {
try {
std::string response_data{};
if (not get_object_list(client, config, response_data, response_code)) {
return false;
}
if (response_code != 200) {
return false;
}
pugi::xml_document doc;
auto res = doc.load_string(response_data.c_str());
if (res.status != pugi::xml_parse_status::status_ok) {
return false;
}
auto node_list = doc.select_nodes("/ListBucketResult/Contents");
for (const auto &node : node_list) {
std::string object_name =
node.node().select_node("Key").node().text().as_string();
if (not utils::string::ends_with(object_name, "/")) {
api_file file{};
object_name = get_name(
*(utils::string::split(object_name, '/', false).end() - 1u),
object_name);
file.api_path = utils::path::create_api_path(object_name);
file.api_parent = utils::path::get_parent_api_path(file.api_path);
file.accessed_date = utils::get_file_time_now();
file.encryption_token = get_api_file_token(file.api_path);
auto size = node.node().select_node("Size").node().text().as_ullong();
file.file_size = file.encryption_token.empty()
? size
: utils::encryption::encrypting_reader::
calculate_decrypted_size(size);
file.changed_date = utils::convert_api_date(
node.node().select_node("LastModified").node().text().as_string());
file.creation_date = file.changed_date;
file.modified_date = file.changed_date;
result.emplace_back(std::move(file));
}
}
return true;
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto list_objects_in_directory_request_impl(
i_http_comm &client, const s3_config &config,
const std::string &object_name, meta_provider_callback meta_provider,
list_objects_result &result, long &response_code) -> bool {
try {
std::string response_data{};
auto prefix = object_name.empty() ? object_name : object_name + "/";
if (not get_object_list(client, config, response_data, response_code, "/",
prefix)) {
return false;
}
if (response_code != 200) {
return false;
}
pugi::xml_document doc;
auto res = doc.load_string(response_data.c_str());
if (res.status != pugi::xml_parse_status::status_ok) {
return false;
}
const auto add_directory_item =
[&](bool directory, const std::string &name,
std::function<std::uint64_t(const directory_item &)> get_size) {
directory_item di{};
di.api_path =
utils::path::create_api_path(utils::path::combine("/", {name}));
di.api_parent = utils::path::get_parent_api_path(di.api_path);
di.directory = directory;
di.size = get_size(di);
meta_provider(di);
result.emplace_back(std::move(di));
};
auto node_list =
doc.select_nodes("/ListBucketResult/CommonPrefixes/Prefix");
for (const auto &node : node_list) {
add_directory_item(
true, node.node().text().as_string(),
[](const directory_item &) -> std::uint64_t { return 0U; });
}
node_list = doc.select_nodes("/ListBucketResult/Contents");
for (const auto &node : node_list) {
auto child_object_name =
node.node().select_node("Key").node().text().as_string();
if (child_object_name != prefix) {
auto size = node.node().select_node("Size").node().text().as_ullong();
add_directory_item(
false, child_object_name,
[&size](const directory_item &) -> std::uint64_t { return size; });
}
}
return true;
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto list_objects_request_impl(i_http_comm &client, const s3_config &config,
list_objects_result &result, long &response_code)
-> bool {
try {
std::string response_data{};
if (not get_object_list(client, config, response_data, response_code)) {
return false;
}
if (response_code != 200) {
return false;
}
pugi::xml_document doc;
auto res = doc.load_string(response_data.c_str());
if (res.status != pugi::xml_parse_status::status_ok) {
return false;
}
auto node_list = doc.select_nodes("/ListBucketResult/Contents");
for (const auto &node : node_list) {
auto object_name =
node.node().select_node("Key").node().text().as_string();
auto size = node.node().select_node("Size").node().text().as_ullong();
directory_item di{};
di.api_path = utils::path::create_api_path(object_name);
di.api_parent = utils::path::get_parent_api_path(di.api_path);
di.directory = utils::string::ends_with(object_name, "/");
di.size = di.directory ? 0U : size;
di.resolved = false;
result.emplace_back(std::move(di));
}
return true;
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto put_object_request_impl(i_http_comm &client, const s3_config &config,
std::string object_name,
const std::string &source_path,
const std::string &encryption_token,
get_key_callback get_key, set_key_callback set_key,
long &response_code, stop_type &stop_requested)
-> bool {
try {
curl::requests::http_put_file put_file{};
put_file.aws_service = "aws:amz:" + config.region + ":s3";
put_file.encryption_token = encryption_token;
put_file.file_name =
*(utils::string::split(object_name, '/', false).end() - 1U);
put_file.path = '/' + object_name;
put_file.source_path = source_path;
if (not encryption_token.empty()) {
static stop_type no_stop{false};
put_file.reader = std::make_shared<utils::encryption::encrypting_reader>(
put_file.file_name, source_path, no_stop, encryption_token,
std::nullopt, -1);
auto key = get_key();
if (key.empty()) {
key = put_file.reader->get_encrypted_file_name();
set_key(key);
}
}
return client.make_request(put_file, response_code, stop_requested);
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
auto read_object_request_impl(i_http_comm &client, const s3_config &config,
const std::string &object_name, std::size_t size,
std::uint64_t offset, data_buffer &data,
long &response_code, stop_type &stop_requested)
-> bool {
try {
curl::requests::http_get get{};
get.aws_service = "aws:amz:" + config.region + ":s3";
get.headers["response-content-type"] = "binary/octet-stream";
get.path = '/' + object_name;
get.range = {{offset, offset + size - 1}};
get.response_handler = [&data](const data_buffer &response_data,
long /*response_code*/) {
data = response_data;
};
return client.make_request(get, response_code, stop_requested);
} catch (const std::exception &e) {
utils::error::raise_error(__FUNCTION__, e, "exception occurred");
}
return false;
}
} // namespace repertory
#endif