mirror of
https://github.com/OpenSpace/OpenSpace.git
synced 2026-01-06 03:29:44 -06:00
@@ -24,17 +24,10 @@
|
||||
|
||||
#include <modules/sync/syncs/httpsynchronization.h>
|
||||
|
||||
#include <modules/sync/syncmodule.h>
|
||||
#include <openspace/documentation/documentation.h>
|
||||
#include <openspace/documentation/verifier.h>
|
||||
#include <openspace/util/httprequest.h>
|
||||
#include <ghoul/fmt.h>
|
||||
#include <ghoul/filesystem/file.h>
|
||||
#include <ghoul/filesystem/filesystem.h>
|
||||
#include <ghoul/logging/logmanager.h>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <numeric>
|
||||
|
||||
namespace {
|
||||
constexpr const char* _loggerCat = "HttpSynchronization";
|
||||
@@ -47,10 +40,11 @@ namespace {
|
||||
constexpr const int ApplicationVersion = 1;
|
||||
|
||||
struct [[codegen::Dictionary(HttpSynchronization)]] Parameters {
|
||||
// A unique identifier for this resource
|
||||
// The unique identifier for this resource that is used to request a set of files
|
||||
// from the synchronization servers
|
||||
std::string identifier;
|
||||
|
||||
// The version of this resource
|
||||
// The version of this resource that should be requested
|
||||
int version;
|
||||
};
|
||||
#include "httpsynchronization_codegen.cpp"
|
||||
@@ -63,12 +57,11 @@ documentation::Documentation HttpSynchronization::Documentation() {
|
||||
}
|
||||
|
||||
HttpSynchronization::HttpSynchronization(const ghoul::Dictionary& dict,
|
||||
std::string synchronizationRoot,
|
||||
std::filesystem::path synchronizationRoot,
|
||||
std::vector<std::string> synchronizationRepositories
|
||||
)
|
||||
: openspace::ResourceSynchronization(dict)
|
||||
, _synchronizationRoot(std::move(synchronizationRoot))
|
||||
, _synchronizationRepositories(std::move(synchronizationRepositories))
|
||||
: ResourceSynchronization(std::move(synchronizationRoot))
|
||||
, _syncRepositories(std::move(synchronizationRepositories))
|
||||
{
|
||||
const Parameters p = codegen::bake<Parameters>(dict);
|
||||
|
||||
@@ -83,40 +76,38 @@ HttpSynchronization::~HttpSynchronization() {
|
||||
}
|
||||
}
|
||||
|
||||
std::string HttpSynchronization::directory() {
|
||||
std::string d = fmt::format(
|
||||
"{}/http/{}/{}", _synchronizationRoot, _identifier, _version
|
||||
);
|
||||
return absPath(d).string();
|
||||
std::filesystem::path HttpSynchronization::directory() const {
|
||||
return _synchronizationRoot / "http" / _identifier / std::to_string(_version);
|
||||
}
|
||||
|
||||
void HttpSynchronization::start() {
|
||||
if (isSyncing()) {
|
||||
return;
|
||||
}
|
||||
begin();
|
||||
_state = State::Syncing;
|
||||
|
||||
if (hasSyncFile()) {
|
||||
resolve();
|
||||
_state = State::Resolved;
|
||||
return;
|
||||
}
|
||||
|
||||
const std::string& query =
|
||||
std::string("?") + QueryKeyIdentifier + "=" + _identifier + "&" +
|
||||
QueryKeyFileVersion + "=" + std::to_string(_version) + "&" +
|
||||
QueryKeyApplicationVersion + "=" + std::to_string(ApplicationVersion);
|
||||
std::string query = fmt::format(
|
||||
"?identifier={}&file_version={}&application_version={}",
|
||||
_identifier, _version, ApplicationVersion
|
||||
);
|
||||
|
||||
_syncThread = std::thread(
|
||||
[this](const std::string& q) {
|
||||
for (const std::string& url : _synchronizationRepositories) {
|
||||
if (trySyncFromUrl(url + q)) {
|
||||
for (const std::string& url : _syncRepositories) {
|
||||
const bool success = trySyncFromUrl(url + q);
|
||||
if (success) {
|
||||
createSyncFile();
|
||||
resolve();
|
||||
_state = State::Resolved;
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (!_shouldCancel) {
|
||||
reject();
|
||||
_state = State::Rejected;
|
||||
}
|
||||
},
|
||||
query
|
||||
@@ -125,174 +116,131 @@ void HttpSynchronization::start() {
|
||||
|
||||
void HttpSynchronization::cancel() {
|
||||
_shouldCancel = true;
|
||||
reset();
|
||||
}
|
||||
|
||||
void HttpSynchronization::clear() {
|
||||
cancel();
|
||||
// TODO: Remove all files from directory.
|
||||
}
|
||||
|
||||
size_t HttpSynchronization::nSynchronizedBytes() {
|
||||
return _nSynchronizedBytes;
|
||||
}
|
||||
|
||||
size_t HttpSynchronization::nTotalBytes() {
|
||||
return _nTotalBytes;
|
||||
}
|
||||
|
||||
bool HttpSynchronization::nTotalBytesIsKnown() {
|
||||
return _nTotalBytesKnown;
|
||||
}
|
||||
|
||||
void HttpSynchronization::createSyncFile() {
|
||||
const std::string& directoryName = directory();
|
||||
const std::string& filepath = directoryName + ".ossync";
|
||||
|
||||
std::filesystem::create_directories(directoryName);
|
||||
|
||||
std::ofstream syncFile(filepath, std::ofstream::out);
|
||||
syncFile << "Synchronized";
|
||||
syncFile.close();
|
||||
}
|
||||
|
||||
bool HttpSynchronization::hasSyncFile() {
|
||||
const std::string& path = directory() + ".ossync";
|
||||
return std::filesystem::is_regular_file(path);
|
||||
_state = State::Unsynced;
|
||||
}
|
||||
|
||||
bool HttpSynchronization::trySyncFromUrl(std::string listUrl) {
|
||||
HttpRequest::RequestOptions opt = {};
|
||||
opt.requestTimeoutSeconds = 0;
|
||||
|
||||
SyncHttpMemoryDownload fileListDownload(std::move(listUrl));
|
||||
fileListDownload.onProgress([&c = _shouldCancel](HttpRequest::Progress) {
|
||||
HttpMemoryDownload fileListDownload(std::move(listUrl));
|
||||
fileListDownload.onProgress([&c = _shouldCancel](int64_t, std::optional<int64_t>) {
|
||||
return !c;
|
||||
});
|
||||
fileListDownload.download(opt);
|
||||
fileListDownload.start();
|
||||
const bool success = fileListDownload.wait();
|
||||
|
||||
if (!fileListDownload.hasSucceeded()) {
|
||||
const std::vector<char>& buffer = fileListDownload.downloadedData();
|
||||
if (!success) {
|
||||
LERRORC("HttpSynchronization", std::string(buffer.begin(), buffer.end()));
|
||||
return false;
|
||||
}
|
||||
|
||||
const std::vector<char>& buffer = fileListDownload.downloadedData();
|
||||
_nSynchronizedBytes = 0;
|
||||
_nTotalBytes = 0;
|
||||
_nTotalBytesKnown = false;
|
||||
|
||||
std::istringstream fileList(std::string(buffer.begin(), buffer.end()));
|
||||
|
||||
std::string line;
|
||||
|
||||
struct SizeData {
|
||||
bool totalKnown;
|
||||
size_t totalBytes;
|
||||
size_t downloadedBytes;
|
||||
int64_t downloadedBytes = 0;
|
||||
std::optional<int64_t> totalBytes;
|
||||
};
|
||||
|
||||
std::unordered_map<std::string, SizeData> sizeData;
|
||||
std::mutex sizeDataMutex;
|
||||
std::mutex mutex;
|
||||
|
||||
std::atomic_bool startedAllDownloads(false);
|
||||
std::atomic_bool startedAllDownloads = false;
|
||||
|
||||
std::vector<std::unique_ptr<AsyncHttpFileDownload>> downloads;
|
||||
// Yes, it should be possible to store this in a std::vector<HttpFileDownload> but
|
||||
// C++ really doesn't like that even though all of the move constructors, move
|
||||
// assignments and everything is automatically constructed
|
||||
std::vector<std::unique_ptr<HttpFileDownload>> downloads;
|
||||
|
||||
std::string line;
|
||||
while (fileList >> line) {
|
||||
size_t lastSlash = line.find_last_of('/');
|
||||
std::string filename = line.substr(lastSlash + 1);
|
||||
|
||||
std::string fileDestination = fmt::format(
|
||||
"{}/{}{}", directory(), filename, TempSuffix
|
||||
);
|
||||
|
||||
if (sizeData.find(line) != sizeData.end()) {
|
||||
LWARNING(fmt::format("{}: Duplicate entries: {}", _identifier, line));
|
||||
if (line.empty() || line[0] == '#') {
|
||||
// Skip all empty lines and commented out lines
|
||||
continue;
|
||||
}
|
||||
|
||||
downloads.push_back(std::make_unique<AsyncHttpFileDownload>(
|
||||
line,
|
||||
fileDestination,
|
||||
HttpFileDownload::Overwrite::Yes
|
||||
));
|
||||
std::string filename = std::filesystem::path(line).filename().string();
|
||||
std::filesystem::path destination = directory() / (filename + TempSuffix);
|
||||
|
||||
std::unique_ptr<AsyncHttpFileDownload>& fileDownload = downloads.back();
|
||||
if (sizeData.find(line) != sizeData.end()) {
|
||||
LWARNING(fmt::format("{}: Duplicate entry for {}", _identifier, line));
|
||||
continue;
|
||||
}
|
||||
|
||||
sizeData[line] = { false, 0, 0 };
|
||||
std::unique_ptr<HttpFileDownload> download =
|
||||
std::make_unique<HttpFileDownload>(
|
||||
line,
|
||||
destination,
|
||||
HttpFileDownload::Overwrite::Yes
|
||||
);
|
||||
HttpFileDownload* dl = download.get();
|
||||
downloads.push_back(std::move(download));
|
||||
|
||||
fileDownload->onProgress(
|
||||
[this, line, &sizeData, &sizeDataMutex,
|
||||
&startedAllDownloads](HttpRequest::Progress p)
|
||||
sizeData[line] = SizeData();
|
||||
|
||||
dl->onProgress(
|
||||
[this, line, &sizeData, &mutex, &startedAllDownloads](int64_t downloadedBytes,
|
||||
std::optional<int64_t> totalBytes)
|
||||
{
|
||||
if (!p.totalBytesKnown || !startedAllDownloads) {
|
||||
if (!totalBytes.has_value() || !startedAllDownloads) {
|
||||
return !_shouldCancel;
|
||||
}
|
||||
|
||||
std::lock_guard guard(sizeDataMutex);
|
||||
std::lock_guard guard(mutex);
|
||||
|
||||
sizeData[line] = { p.totalBytesKnown, p.totalBytes, p.downloadedBytes };
|
||||
sizeData[line] = { downloadedBytes, totalBytes };
|
||||
|
||||
SizeData size = std::accumulate(
|
||||
sizeData.begin(),
|
||||
sizeData.end(),
|
||||
SizeData{ true, 0, 0 },
|
||||
[](const SizeData& a, const std::pair<const std::string, SizeData>& b) {
|
||||
return SizeData {
|
||||
a.totalKnown && b.second.totalKnown,
|
||||
a.totalBytes + b.second.totalBytes,
|
||||
a.downloadedBytes + b.second.downloadedBytes
|
||||
};
|
||||
}
|
||||
);
|
||||
|
||||
_nTotalBytesKnown = size.totalKnown;
|
||||
_nTotalBytes = size.totalBytes;
|
||||
_nSynchronizedBytes = size.downloadedBytes;
|
||||
_nTotalBytesKnown = true;
|
||||
_nTotalBytes = 0;
|
||||
_nSynchronizedBytes = 0;
|
||||
for (const std::pair<const std::string, SizeData>& sd : sizeData) {
|
||||
_nTotalBytesKnown = _nTotalBytesKnown && sd.second.totalBytes.has_value();
|
||||
_nTotalBytes += sd.second.totalBytes.value_or(0);
|
||||
_nSynchronizedBytes += sd.second.downloadedBytes;
|
||||
}
|
||||
|
||||
return !_shouldCancel;
|
||||
});
|
||||
|
||||
fileDownload->start(opt);
|
||||
dl->start();
|
||||
}
|
||||
startedAllDownloads = true;
|
||||
|
||||
bool failed = false;
|
||||
for (std::unique_ptr<AsyncHttpFileDownload>& d : downloads) {
|
||||
for (const std::unique_ptr<HttpFileDownload>& d : downloads) {
|
||||
d->wait();
|
||||
if (d->hasSucceeded()) {
|
||||
// If we are forcing the override, we download to a temporary file
|
||||
// first, so when we are done here, we need to rename the file to the
|
||||
// original name
|
||||
|
||||
const std::string& tempName = d->destination();
|
||||
std::string originalName = tempName.substr(
|
||||
0,
|
||||
tempName.size() - strlen(TempSuffix)
|
||||
);
|
||||
|
||||
if (std::filesystem::is_regular_file(originalName)) {
|
||||
std::filesystem::remove(originalName);
|
||||
}
|
||||
int success = rename(tempName.c_str(), originalName.c_str());
|
||||
if (success != 0) {
|
||||
LERROR(fmt::format(
|
||||
"Error renaming file {} to {}", tempName, originalName
|
||||
));
|
||||
failed = true;
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (!d->hasSucceeded()) {
|
||||
LERROR(fmt::format("Error downloading file from URL {}", d->url()));
|
||||
failed = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
// If we are forcing the override, we download to a temporary file first, so when
|
||||
// we are done here, we need to rename the file to the original name
|
||||
|
||||
std::filesystem::path tempName = d->destination();
|
||||
std::filesystem::path originalName = tempName;
|
||||
// Remove the .tmp extension
|
||||
originalName.replace_extension("");
|
||||
|
||||
if (std::filesystem::is_regular_file(originalName)) {
|
||||
std::filesystem::remove(originalName);
|
||||
}
|
||||
std::error_code ec;
|
||||
std::filesystem::rename(tempName, originalName, ec);
|
||||
if (ec) {
|
||||
LERROR(fmt::format("Error renaming {} to {}", tempName, originalName));
|
||||
failed = true;
|
||||
}
|
||||
}
|
||||
if (!failed) {
|
||||
return true;
|
||||
if (failed) {
|
||||
for (const std::unique_ptr<HttpFileDownload>& d : downloads) {
|
||||
d->cancel();
|
||||
}
|
||||
}
|
||||
for (std::unique_ptr<AsyncHttpFileDownload>& d : downloads) {
|
||||
d->cancel();
|
||||
}
|
||||
return false;
|
||||
return !failed;
|
||||
}
|
||||
|
||||
} // namespace openspace
|
||||
|
||||
@@ -32,40 +32,77 @@
|
||||
|
||||
namespace openspace {
|
||||
|
||||
/**
|
||||
* A concreate ResourceSynchronization that will request a list of files from a central
|
||||
* server (the server list is provided in the constructor) by asking for a specific
|
||||
* identifier and a file version and application version addition. The server is expected
|
||||
* to return a flat list of files that can be then directly downloaded into the #directory
|
||||
* of this synchronization. That list of files can have empty lines and commented out
|
||||
* lines (starting with a #) that will be ignored. Every other line is URL that will be
|
||||
* downloaded into the #directory.
|
||||
* Each requested set of files is identified by a triplet of (identifier, file version,
|
||||
* application version). The identifier is denoting the group of files that is requested,
|
||||
* the file version is the specific version of this set of files, and the application
|
||||
* version is reserved for changes in the data transfer format.
|
||||
*/
|
||||
class HttpSynchronization : public ResourceSynchronization {
|
||||
public:
|
||||
HttpSynchronization(const ghoul::Dictionary& dict, std::string synchronizationRoot,
|
||||
/**
|
||||
* The constructor for this synchronization object. The \p dict contains information
|
||||
* about the \c identifier and the \version (which is the file version), the
|
||||
* \p synchronizationRoot is the path to the root folder where the downloaded files
|
||||
* will be placed, and the \p synchronizationRepositories is a list of the URLs which
|
||||
* will be asked to resolve the (identifier, version) pair. The first URL in the list
|
||||
* that can successfully resolve the requested (identifier, version) pair is the one
|
||||
* that will be used.
|
||||
*
|
||||
* \param dict The parameter dictionary (namely the identifier and version)
|
||||
* \param synchronizationRoot The path to the root from which the complete #directory
|
||||
* path is constructed
|
||||
* \param synchronizationRepositories The list of repositories that will be asked to
|
||||
* resolve the identifier request
|
||||
*/
|
||||
HttpSynchronization(const ghoul::Dictionary& dict,
|
||||
std::filesystem::path synchronizationRoot,
|
||||
std::vector<std::string> synchronizationRepositories);
|
||||
|
||||
/// Destructor that will close the asynchronous file transfer, if it is still ongoing
|
||||
virtual ~HttpSynchronization();
|
||||
|
||||
/**
|
||||
* Returns the location to which files downloaded through this ResourceSynchronization
|
||||
* are saved.
|
||||
*
|
||||
* \return The location for files created by this class
|
||||
*/
|
||||
std::filesystem::path directory() const override;
|
||||
|
||||
std::string directory() override;
|
||||
/**
|
||||
* Starts the synchronization for this ResourceSynchronization by first trying to find
|
||||
* a synchronization respository that replies to the request, parsing the result and
|
||||
* then downloading each of the files that are provided in that result.
|
||||
*/
|
||||
void start() override;
|
||||
void cancel() override;
|
||||
void clear() override;
|
||||
|
||||
size_t nSynchronizedBytes() override;
|
||||
size_t nTotalBytes() override;
|
||||
bool nTotalBytesIsKnown() override;
|
||||
/// Cancels any ongoing synchronization of this ResourceSynchronization
|
||||
void cancel() override;
|
||||
|
||||
static documentation::Documentation Documentation();
|
||||
|
||||
private:
|
||||
void createSyncFile();
|
||||
bool hasSyncFile();
|
||||
/// Tries to get a reply from the provided URL and returns that success to the caller
|
||||
bool trySyncFromUrl(std::string url);
|
||||
|
||||
std::atomic_bool _nTotalBytesKnown = false;
|
||||
std::atomic_size_t _nTotalBytes = 0;
|
||||
std::atomic_size_t _nSynchronizedBytes = 0;
|
||||
/// Contains a flag whether the current transfer should be cancelled
|
||||
std::atomic_bool _shouldCancel = false;
|
||||
|
||||
std::string _identifier;
|
||||
/// The file version for the requested files
|
||||
int _version = -1;
|
||||
std::string _synchronizationRoot;
|
||||
std::vector<std::string> _synchronizationRepositories;
|
||||
|
||||
// The list of all repositories that we'll try to sync from
|
||||
const std::vector<std::string> _syncRepositories;
|
||||
|
||||
// The thread that will be doing the synchronization
|
||||
std::thread _syncThread;
|
||||
};
|
||||
|
||||
|
||||
@@ -24,20 +24,12 @@
|
||||
|
||||
#include <modules/sync/syncs/urlsynchronization.h>
|
||||
|
||||
#include <modules/sync/syncmodule.h>
|
||||
#include <openspace/documentation/documentation.h>
|
||||
#include <openspace/documentation/verifier.h>
|
||||
#include <openspace/engine/moduleengine.h>
|
||||
#include <openspace/util/httprequest.h>
|
||||
#include <ghoul/fmt.h>
|
||||
#include <ghoul/logging/logmanager.h>
|
||||
#include <ghoul/filesystem/file.h>
|
||||
#include <ghoul/filesystem/filesystem.h>
|
||||
#include <ghoul/misc/dictionary.h>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <numeric>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <optional>
|
||||
#include <variant>
|
||||
|
||||
@@ -46,14 +38,13 @@ namespace {
|
||||
|
||||
struct [[codegen::Dictionary(UrlSynchronization)]] Parameters {
|
||||
// The URL or urls from where the files are downloaded. If multiple URLs are
|
||||
// provided, all files will be downloaded to the same directory
|
||||
// provided, all files will be downloaded to the same directory and the filename
|
||||
// parameter must not be specified simultaneously
|
||||
std::variant<std::string, std::vector<std::string>> url;
|
||||
|
||||
// This optional identifier will be part of the used folder structure and, if
|
||||
// provided, can be used to manually find the downloaded folder in the
|
||||
// synchronization folder. If this value is not specified, 'UseHash' has to be set
|
||||
// to 'true'
|
||||
std::optional<std::string> identifier;
|
||||
// This identifier will be part of the used folder structure and, can be used to
|
||||
// manually find the downloaded folder in the synchronization folder
|
||||
std::string identifier;
|
||||
|
||||
// If this value is set to 'true' and it is not overwritten by the global
|
||||
// settings, the file(s) pointed to by this URLSynchronization will always be
|
||||
@@ -69,7 +60,8 @@ namespace {
|
||||
std::optional<bool> useHash;
|
||||
|
||||
// Optional to provide filename to override the one which is otherwise
|
||||
// automatically created from the url
|
||||
// automatically created from the url. If this value is specified, the url
|
||||
// parameter only only contain exactly one URL
|
||||
std::optional<std::string> filename;
|
||||
};
|
||||
#include "urlsynchronization_codegen.cpp"
|
||||
@@ -81,12 +73,11 @@ documentation::Documentation UrlSynchronization::Documentation() {
|
||||
return codegen::doc<Parameters>("sync_synchronization_url");
|
||||
}
|
||||
|
||||
UrlSynchronization::UrlSynchronization(const ghoul::Dictionary& dict,
|
||||
std::string synchronizationRoot)
|
||||
: ResourceSynchronization(dict)
|
||||
, _synchronizationRoot(std::move(synchronizationRoot))
|
||||
UrlSynchronization::UrlSynchronization(const ghoul::Dictionary& dictionary,
|
||||
std::filesystem::path synchronizationRoot)
|
||||
: ResourceSynchronization(std::move(synchronizationRoot))
|
||||
{
|
||||
const Parameters p = codegen::bake<Parameters>(dict);
|
||||
const Parameters p = codegen::bake<Parameters>(dictionary);
|
||||
|
||||
if (std::holds_alternative<std::string>(p.url)) {
|
||||
_urls.push_back(std::get<std::string>(p.url));
|
||||
@@ -99,38 +90,30 @@ UrlSynchronization::UrlSynchronization(const ghoul::Dictionary& dict,
|
||||
throw ghoul::MissingCaseException();
|
||||
}
|
||||
|
||||
if (p.filename.has_value() && _urls.size() > 1) {
|
||||
throw ghoul::RuntimeError(fmt::format(
|
||||
"UrlSynchronization ({}) requested overwrite filename but specified {} URLs "
|
||||
"to download, which is not legal",
|
||||
p.identifier, _urls.size()
|
||||
));
|
||||
}
|
||||
_filename = p.filename.value_or(_filename);
|
||||
|
||||
bool useHash = p.useHash.value_or(true);
|
||||
|
||||
// We just merge all of the URLs together to generate a hash, it's not as stable to
|
||||
// reordering URLs, but every other solution would be more error prone
|
||||
std::string urlConcat = std::accumulate(_urls.begin(), _urls.end(), std::string());
|
||||
size_t hash = std::hash<std::string>{}(urlConcat);
|
||||
if (p.identifier.has_value()) {
|
||||
if (useHash) {
|
||||
_identifier = *p.identifier + "(" + std::to_string(hash) + ")";
|
||||
}
|
||||
else {
|
||||
_identifier = *p.identifier;
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (useHash) {
|
||||
_identifier = std::to_string(hash);
|
||||
}
|
||||
else {
|
||||
documentation::TestResult res;
|
||||
res.success = false;
|
||||
documentation::TestResult::Offense o;
|
||||
o.offender = "Identifier|UseHash";
|
||||
o.reason = documentation::TestResult::Offense::Reason::MissingKey;
|
||||
res.offenses.push_back(o);
|
||||
throw documentation::SpecificationError(std::move(res), "UrlSynchronization");
|
||||
}
|
||||
}
|
||||
|
||||
_forceOverride = p.forceOverride.value_or(_forceOverride);
|
||||
|
||||
const bool useHash = p.useHash.value_or(true);
|
||||
|
||||
_identifier = p.identifier;
|
||||
|
||||
if (useHash) {
|
||||
// We just merge all of the URLs together to generate a hash that works for this
|
||||
std::vector<std::string> urls = _urls;
|
||||
std::sort(urls.begin(), urls.end());
|
||||
|
||||
size_t hash = std::hash<std::string>{}(
|
||||
std::accumulate(urls.begin(), urls.end(), std::string())
|
||||
);
|
||||
_identifier += fmt::format("({})", hash);
|
||||
}
|
||||
}
|
||||
|
||||
UrlSynchronization::~UrlSynchronization() {
|
||||
@@ -140,115 +123,107 @@ UrlSynchronization::~UrlSynchronization() {
|
||||
}
|
||||
}
|
||||
|
||||
std::filesystem::path UrlSynchronization::directory() const {
|
||||
return _synchronizationRoot / "url" / _identifier / "files";
|
||||
}
|
||||
|
||||
void UrlSynchronization::start() {
|
||||
if (isSyncing()) {
|
||||
return;
|
||||
}
|
||||
begin();
|
||||
_state = State::Syncing;
|
||||
|
||||
if (hasSyncFile() && !_forceOverride) {
|
||||
resolve();
|
||||
_state = State::Resolved;
|
||||
return;
|
||||
}
|
||||
|
||||
_syncThread = std::thread([this] {
|
||||
_syncThread = std::thread([this]() {
|
||||
std::unordered_map<std::string, size_t> fileSizes;
|
||||
std::mutex fileSizeMutex;
|
||||
std::atomic_size_t nDownloads(0);
|
||||
std::atomic_bool startedAllDownloads(false);
|
||||
std::vector<std::unique_ptr<AsyncHttpFileDownload>> downloads;
|
||||
size_t nDownloads = 0;
|
||||
std::atomic_bool startedAllDownloads = false;
|
||||
std::vector<std::unique_ptr<HttpFileDownload>> downloads;
|
||||
|
||||
for (const std::string& url : _urls) {
|
||||
if (_filename.empty()) {
|
||||
const size_t lastSlash = url.find_last_of('/');
|
||||
std::string lastPartOfUrl = url.substr(lastSlash + 1);
|
||||
std::string name = std::filesystem::path(url).filename().string();
|
||||
|
||||
// We can not create filenames with questionmarks
|
||||
lastPartOfUrl.erase(
|
||||
std::remove(lastPartOfUrl.begin(), lastPartOfUrl.end(), '?'),
|
||||
lastPartOfUrl.end()
|
||||
);
|
||||
_filename = lastPartOfUrl;
|
||||
// We can not create filenames with question marks
|
||||
name.erase(std::remove(name.begin(), name.end(), '?'), name.end());
|
||||
_filename = name;
|
||||
}
|
||||
std::string fileDestination = fmt::format(
|
||||
"{}/{}{}", directory(), _filename, TempSuffix
|
||||
);
|
||||
std::filesystem::path destination = directory() / (_filename + TempSuffix);
|
||||
|
||||
std::unique_ptr<AsyncHttpFileDownload> download =
|
||||
std::make_unique<AsyncHttpFileDownload>(
|
||||
std::unique_ptr<HttpFileDownload> download =
|
||||
std::make_unique<HttpFileDownload>(
|
||||
url,
|
||||
fileDestination,
|
||||
destination,
|
||||
HttpFileDownload::Overwrite::Yes
|
||||
);
|
||||
HttpFileDownload* dl = download.get();
|
||||
|
||||
downloads.push_back(std::move(download));
|
||||
|
||||
std::unique_ptr<AsyncHttpFileDownload>& fileDownload = downloads.back();
|
||||
|
||||
++nDownloads;
|
||||
|
||||
fileDownload->onProgress(
|
||||
dl->onProgress(
|
||||
[this, url, &fileSizes, &fileSizeMutex,
|
||||
&startedAllDownloads, &nDownloads](HttpRequest::Progress p)
|
||||
&startedAllDownloads, &nDownloads](int64_t,
|
||||
std::optional<int64_t> totalBytes)
|
||||
{
|
||||
if (p.totalBytesKnown) {
|
||||
std::lock_guard guard(fileSizeMutex);
|
||||
fileSizes[url] = p.totalBytes;
|
||||
if (!totalBytes.has_value()) {
|
||||
return !_shouldCancel;
|
||||
}
|
||||
|
||||
if (!_nTotalBytesKnown && startedAllDownloads &&
|
||||
fileSizes.size() == nDownloads)
|
||||
{
|
||||
_nTotalBytesKnown = true;
|
||||
_nTotalBytes = std::accumulate(
|
||||
fileSizes.begin(),
|
||||
fileSizes.end(),
|
||||
size_t(0),
|
||||
[](size_t a, const std::pair<const std::string, size_t> b) {
|
||||
return a + b.second;
|
||||
}
|
||||
);
|
||||
std::lock_guard guard(fileSizeMutex);
|
||||
fileSizes[url] = *totalBytes;
|
||||
|
||||
if (!_nTotalBytesKnown && startedAllDownloads &&
|
||||
fileSizes.size() == nDownloads)
|
||||
{
|
||||
_nTotalBytesKnown = true;
|
||||
_nTotalBytes = 0;
|
||||
for (const std::pair<const std::string, size_t>& fs : fileSizes) {
|
||||
_nTotalBytes += fs.second;
|
||||
}
|
||||
}
|
||||
return !_shouldCancel;
|
||||
});
|
||||
|
||||
HttpRequest::RequestOptions opt = {};
|
||||
opt.requestTimeoutSeconds = 0;
|
||||
fileDownload->start(opt);
|
||||
dl->start();
|
||||
}
|
||||
|
||||
startedAllDownloads = true;
|
||||
|
||||
bool failed = false;
|
||||
for (std::unique_ptr<AsyncHttpFileDownload>& d : downloads) {
|
||||
for (const std::unique_ptr<HttpFileDownload>& d : downloads) {
|
||||
d->wait();
|
||||
if (d->hasSucceeded()) {
|
||||
// If we are forcing the override, we download to a temporary file first,
|
||||
// so when we are done here, we need to rename the file to the original
|
||||
// name
|
||||
if (!d->hasSucceeded()) {
|
||||
failed = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
const std::string& tempName = d->destination();
|
||||
std::string originalName = tempName.substr(
|
||||
0,
|
||||
tempName.size() - strlen(TempSuffix)
|
||||
// If we are forcing the override, we download to a temporary file first, so
|
||||
// when we are done here, we need to rename the file to the original name
|
||||
|
||||
std::filesystem::path tempName = d->destination();
|
||||
std::filesystem::path originalName = tempName;
|
||||
// Remove the .tmp extension
|
||||
originalName.replace_extension("");
|
||||
|
||||
if (std::filesystem::is_regular_file(originalName)) {
|
||||
std::filesystem::remove(originalName);
|
||||
}
|
||||
|
||||
std::error_code ec;
|
||||
std::filesystem::rename(tempName, originalName, ec);
|
||||
if (ec) {
|
||||
LERRORC(
|
||||
"URLSynchronization",
|
||||
fmt::format("Error renaming file {} to {}", tempName, originalName)
|
||||
);
|
||||
|
||||
if (std::filesystem::is_regular_file(originalName)) {
|
||||
std::filesystem::remove(originalName);
|
||||
}
|
||||
int success = rename(tempName.c_str(), originalName.c_str());
|
||||
if (success != 0) {
|
||||
LERRORC(
|
||||
"URLSynchronization",
|
||||
fmt::format(
|
||||
"Error renaming file {} to {}", tempName, originalName
|
||||
)
|
||||
);
|
||||
|
||||
failed = true;
|
||||
}
|
||||
}
|
||||
else {
|
||||
failed = true;
|
||||
}
|
||||
}
|
||||
@@ -257,53 +232,17 @@ void UrlSynchronization::start() {
|
||||
createSyncFile();
|
||||
}
|
||||
else {
|
||||
for (std::unique_ptr<AsyncHttpFileDownload>& d : downloads) {
|
||||
for (const std::unique_ptr<HttpFileDownload>& d : downloads) {
|
||||
d->cancel();
|
||||
}
|
||||
}
|
||||
resolve();
|
||||
_state = State::Resolved;
|
||||
});
|
||||
}
|
||||
|
||||
void UrlSynchronization::cancel() {
|
||||
_shouldCancel = true;
|
||||
reset();
|
||||
}
|
||||
|
||||
void UrlSynchronization::clear() {
|
||||
cancel();
|
||||
// TODO: Remove all files from directory.
|
||||
}
|
||||
|
||||
size_t UrlSynchronization::nSynchronizedBytes() {
|
||||
return _nSynchronizedBytes;
|
||||
}
|
||||
|
||||
size_t UrlSynchronization::nTotalBytes() {
|
||||
return _nTotalBytes;
|
||||
}
|
||||
|
||||
bool UrlSynchronization::nTotalBytesIsKnown() {
|
||||
return _nTotalBytesKnown;
|
||||
}
|
||||
|
||||
void UrlSynchronization::createSyncFile() {
|
||||
std::string dir = directory();
|
||||
std::string filepath = dir + ".ossync";
|
||||
std::filesystem::create_directories(dir);
|
||||
std::ofstream syncFile(filepath, std::ofstream::out);
|
||||
syncFile << "Synchronized";
|
||||
syncFile.close();
|
||||
}
|
||||
|
||||
bool UrlSynchronization::hasSyncFile() {
|
||||
const std::string& path = directory() + ".ossync";
|
||||
return std::filesystem::is_regular_file(path);
|
||||
}
|
||||
|
||||
std::string UrlSynchronization::directory() {
|
||||
std::string d = fmt::format("{}/url/{}/files", _synchronizationRoot, _identifier);
|
||||
return absPath(d).string();
|
||||
_state = State::Unsynced;
|
||||
}
|
||||
|
||||
} // namespace openspace
|
||||
|
||||
@@ -28,43 +28,68 @@
|
||||
#include <openspace/util/resourcesynchronization.h>
|
||||
|
||||
#include <atomic>
|
||||
#include <filesystem>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
namespace openspace {
|
||||
|
||||
/**
|
||||
* The UrlSynchronization will download one or more files by directly being provided with
|
||||
* the list of URLs to the files that should be downloaded. The \c Override option in the
|
||||
* Dictionary determines what should happen in a file with the same name and the same
|
||||
* identifier has been previously downloaded.
|
||||
*/
|
||||
class UrlSynchronization : public ResourceSynchronization {
|
||||
public:
|
||||
UrlSynchronization(const ghoul::Dictionary& dict, std::string synchronizationRoot);
|
||||
/**
|
||||
* The constructor that takes the parameter \p dictionary and the
|
||||
* \p synchronizationRoot to the location that is used as the base to compute the
|
||||
* final storage location. The provided list of URLs must not contain any duplicates.
|
||||
*
|
||||
* \param dictionary The parameter dictionary that contains all information that this
|
||||
* UrlSynchronization needs to download the provided files
|
||||
* \param synchronizationRoot The base location based off which the final placement
|
||||
* is calculated
|
||||
*/
|
||||
UrlSynchronization(const ghoul::Dictionary& dictionary,
|
||||
std::filesystem::path synchronizationRoot);
|
||||
|
||||
/// Contructor that will terminate the synchronization thread if it is still running
|
||||
virtual ~UrlSynchronization();
|
||||
|
||||
void start() override;
|
||||
void cancel() override;
|
||||
void clear() override;
|
||||
/**
|
||||
* Returns the location to which files downloaded through this ResourceSynchronization
|
||||
* are saved.
|
||||
*
|
||||
* \return The location for files created by this class
|
||||
*/
|
||||
std::filesystem::path directory() const override;
|
||||
|
||||
size_t nSynchronizedBytes() override;
|
||||
size_t nTotalBytes() override;
|
||||
bool nTotalBytesIsKnown() override;
|
||||
/// Starts the synchronization for this ResourceSynchronization
|
||||
void start() override;
|
||||
|
||||
/// Cancels any ongoing synchronization of this ResourceSynchronization
|
||||
void cancel() override;
|
||||
|
||||
static documentation::Documentation Documentation();
|
||||
|
||||
private:
|
||||
void createSyncFile();
|
||||
bool hasSyncFile();
|
||||
std::string directory() override;
|
||||
|
||||
/// The list of URLs that will be downloaded
|
||||
std::vector<std::string> _urls;
|
||||
|
||||
/// Setting whether existing files should be ignored (false) or overwritten (true)
|
||||
bool _forceOverride = false;
|
||||
std::string _synchronizationRoot;
|
||||
std::string _identifier;
|
||||
|
||||
/// An optional filename that might overwrite the storage destination. This is only
|
||||
/// valid if a single URL is specified
|
||||
std::string _filename;
|
||||
|
||||
std::atomic_bool _nTotalBytesKnown = false;
|
||||
std::atomic_size_t _nTotalBytes = 0;
|
||||
std::atomic_size_t _nSynchronizedBytes = 0;
|
||||
/// Contains a flag whether the current transfer should be cancelled
|
||||
std::atomic_bool _shouldCancel = false;
|
||||
|
||||
// The thread that will be doing the synchronization
|
||||
std::thread _syncThread;
|
||||
};
|
||||
|
||||
|
||||
Reference in New Issue
Block a user