blakeridgway/supercell-wx
1
0
Fork 0
mirror of https://github.com/ciphervance/supercell-wx.git synced 2025-10-30 20:00:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Additional level 2 data provider functionality for display of latest data

Browse source
This commit is contained in:
Dan Paulat 2022-05-28 02:04:59 -05:00
parent 57e5badd70
commit 4b9d12a7ef
3 changed files with 147 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

130
wxdata/source/scwx/provider/aws_level2_data_provider.cpp
View file

@ -27,6 +27,20 @@ static const std::string kDefaultRegion_ = "us-east-1";
class AwsLevel2DataProviderImpl
{
public:
struct ObjectRecord
{
explicit ObjectRecord(
const std::string& key,
std::chrono::system_clock::time_point lastModified) :
key_ {key}, lastModified_ {lastModified}
{
}
~ObjectRecord() = default;
std::string key_;
std::chrono::system_clock::time_point lastModified_;
};
explicit AwsLevel2DataProviderImpl(const std::string& radarSite,
const std::string& bucketName,
const std::string& region) :
@ -35,7 +49,9 @@ public:
region_ {region},
client_ {nullptr},
objects_ {},
objectsMutex_ {}
objectsMutex_ {},
lastModified_ {},
updatePeriod_ {}
{
Aws::Client::ClientConfiguration config;
config.region = region_;
@ -45,14 +61,19 @@ public:
~AwsLevel2DataProviderImpl() {}
void UpdateMetadata();
std::string radarSite_;
std::string bucketName_;
std::string region_;
std::unique_ptr<Aws::S3::S3Client> client_;
std::map<std::chrono::system_clock::time_point, std::string> objects_;
std::shared_mutex objectsMutex_;
std::map<std::chrono::system_clock::time_point, ObjectRecord> objects_;
std::shared_mutex objectsMutex_;
std::chrono::system_clock::time_point lastModified_;
std::chrono::seconds updatePeriod_;
};
AwsLevel2DataProvider::AwsLevel2DataProvider(const std::string& radarSite) :
@ -78,6 +99,17 @@ size_t AwsLevel2DataProvider::cache_size() const
return p->objects_.size();
}
std::chrono::seconds AwsLevel2DataProvider::update_period() const
{
return p->updatePeriod_;
}
std::chrono::system_clock::time_point
AwsLevel2DataProvider::last_modified() const
{
return p->lastModified_;
}
std::string
AwsLevel2DataProvider::FindKey(std::chrono::system_clock::time_point time)
{
@ -87,12 +119,27 @@ AwsLevel2DataProvider::FindKey(std::chrono::system_clock::time_point time)
std::shared_lock lock(p->objectsMutex_);
std::optional<std::string> element =
util::GetBoundedElement(p->objects_, time);
auto element = util::GetBoundedElement(p->objects_, time);
if (element.has_value())
{
key = *element;
key = element->key_;
}
return key;
}
std::string AwsLevel2DataProvider::FindLatestKey()
{
logger_->debug("FindLatestKey()");
std::string key {};
std::shared_lock lock(p->objectsMutex_);
if (!p->objects_.empty())
{
key = p->objects_.crbegin()->second.key_;
}
return key;
@ -122,29 +169,36 @@ AwsLevel2DataProvider::ListObjects(std::chrono::system_clock::time_point date)
logger_->debug("Found {} objects", objects.size());
// Store objects
std::for_each(objects.cbegin(),
objects.cend(),
[&](const Aws::S3::Model::Object& object)
{
std::string key = object.GetKey();
std::for_each( //
objects.cbegin(),
objects.cend(),
[&](const Aws::S3::Model::Object& object)
{
std::string key = object.GetKey();
if (!key.ends_with("_MDM"))
{
auto time = GetTimePointFromKey(key);
if (!key.ends_with("_MDM"))
{
auto time = GetTimePointFromKey(key);
std::unique_lock lock(p->objectsMutex_);
std::chrono::seconds lastModifiedSeconds {
object.GetLastModified().Seconds()};
std::chrono::system_clock::time_point lastModified {
lastModifiedSeconds};
auto [it, inserted] =
p->objects_.insert_or_assign(time, key);
std::unique_lock lock(p->objectsMutex_);
if (inserted)
{
newObjects++;
}
auto [it, inserted] = p->objects_.insert_or_assign(
time,
AwsLevel2DataProviderImpl::ObjectRecord {key, lastModified});
totalObjects++;
}
});
if (inserted)
{
newObjects++;
}
totalObjects++;
}
});
}
else
{
@ -152,6 +206,8 @@ AwsLevel2DataProvider::ListObjects(std::chrono::system_clock::time_point date)
outcome.GetError().GetMessage());
}
p->UpdateMetadata();
return std::make_pair(newObjects, totalObjects);
}
@ -222,6 +278,32 @@ size_t AwsLevel2DataProvider::Refresh()
return totalNewObjects;
}
void AwsLevel2DataProviderImpl::UpdateMetadata()
{
std::shared_lock lock(objectsMutex_);
if (!objects_.empty())
{
lastModified_ = objects_.crbegin()->second.lastModified_;
}
if (objects_.size() >= 2)
{
auto it = objects_.crbegin();
auto lastModified = it->second.lastModified_;
auto prevModified = (++it)->second.lastModified_;
auto delta = lastModified - prevModified;
updatePeriod_ = std::chrono::duration_cast<std::chrono::seconds>(delta);
}
}
std::chrono::system_clock::time_point
AwsLevel2DataProvider::GetTimePointByKey(const std::string& key) const
{
return GetTimePointFromKey(key);
}
std::chrono::system_clock::time_point
AwsLevel2DataProvider::GetTimePointFromKey(const std::string& key)
{