seems to work now
Some checks failed
EIVE/eive-obsw/pipeline/pr-develop There was a failure building this commit
Some checks failed
EIVE/eive-obsw/pipeline/pr-develop There was a failure building this commit
This commit is contained in:
@ -28,6 +28,42 @@ PersistentTmStore::PersistentTmStore(object_id_t objectId, const char* baseDir,
|
||||
calcDiffSeconds(intervalUnit, intervalCount);
|
||||
}
|
||||
|
||||
ReturnValue_t PersistentTmStore::assignAndOrCreateMostRecentFile() {
|
||||
using namespace std::filesystem;
|
||||
for (auto const& file : directory_iterator(basePath)) {
|
||||
if (file.is_directory()) {
|
||||
continue;
|
||||
}
|
||||
auto pathStr = file.path().string();
|
||||
if (pathStr.find(baseName) == std::string::npos) {
|
||||
continue;
|
||||
}
|
||||
unsigned int underscorePos = pathStr.find_last_of('_');
|
||||
std::string stampStr = pathStr.substr(underscorePos + 1);
|
||||
struct tm time {};
|
||||
if (nullptr == strptime(stampStr.c_str(), FILE_DATE_FORMAT, &time)) {
|
||||
sif::error << "PersistentTmStore::assignOrCreateMostRecentFile: Error reading timestamp"
|
||||
<< std::endl;
|
||||
// Delete the file and re-create it.
|
||||
activeFile = std::nullopt;
|
||||
std::filesystem::remove(file.path());
|
||||
break;
|
||||
}
|
||||
time_t fileEpoch = timegm(&time);
|
||||
// There is still a file within the active time window, so re-use that file for new TMs to
|
||||
// store.
|
||||
if (fileEpoch + rolloverDiffSeconds > currentTv.tv_sec) {
|
||||
activeFileTv.tv_sec = fileEpoch;
|
||||
activeFile = file.path();
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (not activeFile.has_value()) {
|
||||
return createMostRecentFile();
|
||||
}
|
||||
return returnvalue::OK;
|
||||
}
|
||||
|
||||
ReturnValue_t PersistentTmStore::handleCommandQueue(StorageManagerIF& ipcStore,
|
||||
TmFunnelBase& tmFunnel) {
|
||||
CommandMessage cmdMessage;
|
||||
@ -51,14 +87,16 @@ ReturnValue_t PersistentTmStore::handleCommandQueue(StorageManagerIF& ipcStore,
|
||||
} else if (cmd == TmStoreMessage::DOWNLINK_STORE_CONTENT_TIME) {
|
||||
store_address_t storeId = TmStoreMessage::getStoreId(&cmdMessage);
|
||||
auto accessor = ipcStore.getData(storeId);
|
||||
if (accessor.second.size() < 8) {
|
||||
return returnvalue::FAILED;
|
||||
}
|
||||
uint32_t dumpFromUnixSeconds;
|
||||
uint32_t dumpUntilUnixSeconds;
|
||||
size_t size = accessor.second.size();
|
||||
size_t size = 8;
|
||||
SerializeAdapter::deSerialize(&dumpFromUnixSeconds, accessor.second.data(), &size,
|
||||
SerializeIF::Endianness::NETWORK);
|
||||
SerializeAdapter::deSerialize(&dumpUntilUnixSeconds, accessor.second.data(), &size,
|
||||
SerializeAdapter::deSerialize(&dumpUntilUnixSeconds, accessor.second.data() + 4, &size,
|
||||
SerializeIF::Endianness::NETWORK);
|
||||
// TODO: TM store missing, and maybe there is a better way to do this?
|
||||
dumpFromUpTo(dumpFromUnixSeconds, dumpUntilUnixSeconds, tmFunnel);
|
||||
}
|
||||
}
|
||||
@ -110,30 +148,32 @@ ReturnValue_t PersistentTmStore::storePacket(PusTmReader& reader) {
|
||||
updateBaseDir();
|
||||
}
|
||||
// It is assumed here that the filesystem is usable.
|
||||
if (not mostRecentFile) {
|
||||
assignAndOrCreateMostRecentFile();
|
||||
if (not activeFile.has_value()) {
|
||||
ReturnValue_t result = assignAndOrCreateMostRecentFile();
|
||||
if (result != returnvalue::OK) {
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
if (currentTv.tv_sec < mostRecentTv.value().tv_sec or
|
||||
currentTv.tv_sec - mostRecentTv.value().tv_sec > static_cast<int>(rolloverDiffSeconds)) {
|
||||
if (file_size(mostRecentFile.value()) + reader.getFullPacketLen() > fileBuf.size()) {
|
||||
if (currentTv.tv_sec < activeFileTv.tv_sec or
|
||||
currentTv.tv_sec - activeFileTv.tv_sec > static_cast<int>(rolloverDiffSeconds)) {
|
||||
if (file_size(activeFile.value()) + reader.getFullPacketLen() > fileBuf.size()) {
|
||||
uint8_t appendedCounter = 1;
|
||||
path rolloverName;
|
||||
while (true) {
|
||||
rolloverName =
|
||||
path(mostRecentFile.value().string() + "." + std::to_string(appendedCounter));
|
||||
rolloverName = path(activeFile.value().string() + "." + std::to_string(appendedCounter));
|
||||
if (not exists(rolloverName)) {
|
||||
break;
|
||||
}
|
||||
appendedCounter++;
|
||||
}
|
||||
rename(mostRecentFile.value(), rolloverName);
|
||||
std::ofstream of(mostRecentFile.value(), std::ios::binary);
|
||||
rename(activeFile.value(), rolloverName);
|
||||
std::ofstream of(activeFile.value(), std::ios::binary);
|
||||
}
|
||||
}
|
||||
|
||||
// Rollover conditions were handled, write to file now
|
||||
std::ofstream of(mostRecentFile.value(), std::ios::app | std::ios::binary);
|
||||
std::ofstream of(activeFile.value(), std::ios::app | std::ios::binary);
|
||||
of.write(reinterpret_cast<const char*>(reader.getFullData()),
|
||||
static_cast<std::streamsize>(reader.getFullPacketLen()));
|
||||
return returnvalue::OK;
|
||||
@ -161,51 +201,6 @@ void PersistentTmStore::updateBaseDir() {
|
||||
baseDirUninitialized = false;
|
||||
}
|
||||
|
||||
void PersistentTmStore::assignAndOrCreateMostRecentFile() {
|
||||
using namespace std::filesystem;
|
||||
for (auto const& file : directory_iterator(basePath)) {
|
||||
if (file.is_directory()) {
|
||||
continue;
|
||||
}
|
||||
auto pathStr = file.path().string();
|
||||
Clock::TimeOfDay_t tod;
|
||||
if (pathStr.find(baseName) == std::string::npos) {
|
||||
continue;
|
||||
}
|
||||
unsigned int underscorePos = pathStr.find_last_of('_');
|
||||
std::string stampStr = pathStr.substr(underscorePos + 1);
|
||||
int count = sscanf(stampStr.c_str(),
|
||||
"%04" SCNu32 "-%02" SCNu32 "-%02" SCNu32 "T%02" SCNu32 "-%02" SCNu32
|
||||
"-%02" SCNu32 "Z",
|
||||
&tod.year, &tod.month, &tod.day, &tod.hour, &tod.minute, &tod.second);
|
||||
if (count != 6) {
|
||||
continue;
|
||||
}
|
||||
timeval tv{};
|
||||
Clock::convertTimeOfDayToTimeval(&tod, &tv);
|
||||
if (not mostRecentTv || tv > mostRecentTv.value()) {
|
||||
mostRecentTv = tv;
|
||||
mostRecentFile = file.path();
|
||||
}
|
||||
}
|
||||
if (not mostRecentFile) {
|
||||
unsigned currentIdx = 0;
|
||||
path pathStart = basePath / baseName;
|
||||
memcpy(fileBuf.data() + currentIdx, pathStart.c_str(), pathStart.string().length());
|
||||
currentIdx += pathStart.string().length();
|
||||
Clock::TimeOfDay_t tod;
|
||||
Clock::convertTimevalToTimeOfDay(¤tTv, &tod);
|
||||
currentIdx += sprintf(reinterpret_cast<char*>(fileBuf.data() + currentIdx),
|
||||
"_%04" PRIu32 "-%02" PRIu32 "-%02" PRIu32 "T%02" PRIu32 "-%02" PRIu32
|
||||
"-%02" PRIu32 "Z.bin",
|
||||
tod.year, tod.month, tod.day, tod.hour, tod.minute, tod.second);
|
||||
path newPath(std::string(reinterpret_cast<const char*>(fileBuf.data()), currentIdx));
|
||||
std::ofstream of(newPath, std::ios::binary);
|
||||
mostRecentFile = newPath;
|
||||
mostRecentTv = currentTv;
|
||||
}
|
||||
}
|
||||
|
||||
void PersistentTmStore::addApid(uint16_t apid) {
|
||||
if (not filter.apid) {
|
||||
filter.apid = std::vector<uint16_t>({apid});
|
||||
@ -234,19 +229,17 @@ void PersistentTmStore::addServiceSubservice(uint8_t service, uint8_t subservice
|
||||
void PersistentTmStore::deleteUpTo(uint32_t unixSeconds) {
|
||||
using namespace std::filesystem;
|
||||
for (auto const& file : directory_iterator(basePath)) {
|
||||
if (file.is_directory() or
|
||||
(mostRecentFile.has_value() and (mostRecentFile.value() == file.path()))) {
|
||||
if (file.is_directory() or (activeFile.has_value() and (activeFile.value() == file.path()))) {
|
||||
continue;
|
||||
}
|
||||
Clock::TimeOfDay_t tod;
|
||||
pathToTod(file.path(), tod);
|
||||
timeval time{};
|
||||
ReturnValue_t result = Clock::convertTimeOfDayToTimeval(&tod, &time);
|
||||
if (result != returnvalue::OK) {
|
||||
sif::error << "TOD to time conversion failed for file " << file << std::endl;
|
||||
// Convert file time to the UNIX epoch
|
||||
struct tm fileTime {};
|
||||
if (pathToTm(file.path(), fileTime) != returnvalue::OK) {
|
||||
sif::error << "Time extraction for " << file << "failed" << std::endl;
|
||||
continue;
|
||||
}
|
||||
if (time.tv_sec + rolloverDiffSeconds < unixSeconds) {
|
||||
time_t epoch = timegm(&fileTime);
|
||||
if (epoch + rolloverDiffSeconds < unixSeconds) {
|
||||
std::filesystem::remove(file.path());
|
||||
}
|
||||
}
|
||||
@ -259,33 +252,26 @@ void PersistentTmStore::dumpFromUpTo(uint32_t fromUnixSeconds, uint32_t upToUnix
|
||||
if (file.is_directory()) {
|
||||
continue;
|
||||
}
|
||||
if (mostRecentFile.has_value() and mostRecentTv.has_value() and
|
||||
(file.path() == mostRecentFile.value()) and
|
||||
(upToUnixSeconds < static_cast<uint32_t>(mostRecentTv.value().tv_sec))) {
|
||||
struct tm fileTime {};
|
||||
if (pathToTm(file.path(), fileTime) != returnvalue::OK) {
|
||||
sif::error << "Time extraction for file " << file << "failed" << std::endl;
|
||||
continue;
|
||||
}
|
||||
Clock::TimeOfDay_t tod;
|
||||
pathToTod(file.path(), tod);
|
||||
timeval time{};
|
||||
ReturnValue_t result = Clock::convertTimeOfDayToTimeval(&tod, &time);
|
||||
if (result != returnvalue::OK) {
|
||||
sif::error << "TOD to time conversion failed for file " << file << std::endl;
|
||||
continue;
|
||||
}
|
||||
auto timeUnsigned = static_cast<uint32_t>(time.tv_sec);
|
||||
if (timeUnsigned > fromUnixSeconds && timeUnsigned + rolloverDiffSeconds < upToUnixSeconds) {
|
||||
fileToPackets(file, timeUnsigned, funnel);
|
||||
auto fileEpoch = static_cast<uint32_t>(timegm(&fileTime));
|
||||
if ((fileEpoch > fromUnixSeconds) and (fileEpoch + rolloverDiffSeconds <= upToUnixSeconds)) {
|
||||
fileToPackets(file, fileEpoch, funnel);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void PersistentTmStore::pathToTod(const std::filesystem::path& path, Clock::TimeOfDay_t& tod) {
|
||||
ReturnValue_t PersistentTmStore::pathToTm(const std::filesystem::path& path, struct tm& time) {
|
||||
auto pathStr = path.string();
|
||||
size_t splitChar = pathStr.find('_');
|
||||
auto timeOnlyStr = pathStr.substr(splitChar);
|
||||
sscanf(timeOnlyStr.data(),
|
||||
"%04" SCNu32 "-%02" SCNu32 "-%02" SCNu32 "T%02" SCNu32 "-%02" SCNu32 "-%02" SCNu32 "Z",
|
||||
&tod.year, &tod.month, &tod.day, &tod.hour, &tod.minute, &tod.second);
|
||||
auto timeOnlyStr = pathStr.substr(splitChar + 1);
|
||||
if (nullptr == strptime(timeOnlyStr.c_str(), FILE_DATE_FORMAT, &time)) {
|
||||
return returnvalue::FAILED;
|
||||
}
|
||||
return returnvalue::OK;
|
||||
}
|
||||
|
||||
void PersistentTmStore::fileToPackets(const std::filesystem::path& path, uint32_t unixStamp,
|
||||
@ -320,3 +306,37 @@ void PersistentTmStore::fileToPackets(const std::filesystem::path& path, uint32_
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ReturnValue_t PersistentTmStore::createMostRecentFile() {
|
||||
using namespace std::filesystem;
|
||||
unsigned currentIdx = 0;
|
||||
path pathStart = basePath / baseName;
|
||||
memcpy(fileBuf.data() + currentIdx, pathStart.c_str(), pathStart.string().length());
|
||||
currentIdx += pathStart.string().length();
|
||||
fileBuf[currentIdx] = '_';
|
||||
currentIdx += 1;
|
||||
time_t epoch = currentTv.tv_sec;
|
||||
struct tm* time = gmtime(&epoch);
|
||||
size_t writtenBytes = strftime(reinterpret_cast<char*>(fileBuf.data() + currentIdx),
|
||||
fileBuf.size(), FILE_DATE_FORMAT, time);
|
||||
if (writtenBytes == 0) {
|
||||
sif::error << "PersistentTmStore::createMostRecentFile: Could not create file timestamp"
|
||||
<< std::endl;
|
||||
return returnvalue::FAILED;
|
||||
}
|
||||
currentIdx += writtenBytes;
|
||||
strncpy(reinterpret_cast<char*>(fileBuf.data() + currentIdx), ".bin",
|
||||
fileBuf.size() - currentIdx);
|
||||
currentIdx += 4;
|
||||
|
||||
path newPath(std::string(reinterpret_cast<const char*>(fileBuf.data()), currentIdx));
|
||||
std::ofstream of(newPath, std::ios::binary);
|
||||
activeFile = newPath;
|
||||
activeFileTv = currentTv;
|
||||
return returnvalue::OK;
|
||||
}
|
||||
|
||||
ReturnValue_t PersistentTmStore::initializeTmStore() {
|
||||
updateBaseDir();
|
||||
return assignAndOrCreateMostRecentFile();
|
||||
}
|
||||
|
Reference in New Issue
Block a user