MessagePlayer.cpp

Go to the documentation of this file.

#include "MessagePlayer.h"
 
namespace cmlabs {
 
MessagePlayer::MessagePlayer() {
    interval = 0;
    maxCount = 0;
    maxSize = 0;
    rotate = false;
    indexData = NULL;
    indexSize = 0;
    currentTriggerName[0] = 0;
    indexFileSize = 0;
 
    curStart = 0;
    curWrite = 0;
    curIndex = 0;
    curCount = 0;
    curSize = 0;
    lastWriteTime = GetTimeNow();
    lastReceiveTime = 0;
    lastReadTime = 0;
    serial = 0;
    nextMsgTime = 0;
}
 
MessagePlayer::~MessagePlayer() {
    writeBuffer();
    if (indexData)
        delete(indexData);
    indexData = NULL;
    indexSize = 0;
}
 
bool MessagePlayer::initRead(const char* root, bool rotate) {
    return initRead(root, 0, rotate);
}
 
bool MessagePlayer::initRead(const char* root, uint32 interval, bool rotate) {
    this->interval = interval;
    this->rotate = rotate;
 
    if (interval) {
        //LogPrint(0, 0, 1, "Starting ReplayCatalog playback, interval: %u", interval);
    }
    else {
        this->interval = 1;
        //LogPrint(0, 0, 1, "Starting ReplayCatalog playback with actual message intervals");
    }
 
    // Find directory location
    this->root = root;
    while (utils::TextEndsWith(this->root.c_str(), "/") || utils::TextEndsWith(this->root.c_str(), "\\")) {
        this->root = this->root.substr(0, this->root.length() - 2);
    }
 
    // Backup directory, if already exists
    utils::FileDetails dirDetails = utils::GetFileDetails((this->root + "/index.bin").c_str());
 
    uint32 n = 0;
    utils::FileDetails altDetails;
    // Find backup location
    std::string altRoot = utils::StringFormat("%s.%u", this->root.c_str(), n);
 
    ReplayIndexEntry* entry;
    ReplayIndexEntry initEntry;
    initEntry.reset();
 
    if (!dirDetails.doesExist) {
        while (!(altDetails = utils::GetFileDetails((altRoot + "/index.bin").c_str())).doesExist && n < 100)
            altRoot = utils::StringFormat("%s.%u", this->root.c_str(), ++n);
        if (!altDetails.doesExist) {
            LogPrint(0, 0, 0, "Cannot find MessagePlayer storage location '%s' or any backup locations", this->root.c_str());
            return false;
        }
        else {
            LogPrint(0, 0, 0, "Couldn't find MessagePlayer storage location '%s', using detected location '%s' instead", this->root.c_str(), altRoot.c_str());
            this->root = altRoot;
        }
    }
    else
        LogPrint(0, 0, 0, "Reading from MessagePlayer storage location '%s'", this->root.c_str());
    indexFile = this->root + "/index.bin";
    metadataFile = this->root + "/metadata.bin";
    metadataTextFile = this->root + "/metadata.text";
    csvFile = this->root + "/data.csv";
 
    if (!(indexData = utils::ReadAFile(indexFile.c_str(), indexFileSize, true))) {
        LogPrint(0, 0, 0, "Cannot read MessagePlayer storage location index file '%s'", indexFile.c_str());
    }
    curCount = indexSize = indexFileSize / sizeof(ReplayIndexEntry);
    entry = (ReplayIndexEntry*)indexData;
    if (!entry->isValid()) {
        LogPrint(0, 0, 0, "Index file '%s' corrupted", indexFile.c_str());
        return false;
    }
    if (!entry->size) {
        LogPrint(0, 0, 0, "MessagePlayer storage index for location '%s' empty", this->root.c_str());
        return false;
    }
    readMetadata();
    return true;
}
 
bool MessagePlayer::initWrite(const char* root, uint32 maxCount, uint64 maxSize) {
    this->maxCount = maxCount;
    this->maxSize = maxSize;
    if (maxCount && maxSize)
        LogPrint(0, 0, 1, "Starting ReplayCatalog recording, maxcount: %u and maxsize: %s", maxCount, utils::BytifySize((double)maxSize).c_str());
    else if (maxCount)
        LogPrint(0, 0, 1, "Starting ReplayCatalog recording, maxcount: %u", maxCount);
    else
        LogPrint(0, 0, 1, "Starting ReplayCatalog recording, maxsize: %s", utils::BytifySize((double)maxSize).c_str());
 
    // Find directory location
    this->root = root;
    while (utils::TextEndsWith(this->root.c_str(), "/") || utils::TextEndsWith(this->root.c_str(), "\\")) {
        this->root = this->root.substr(0, this->root.length() - 2);
    }
 
    // Backup directory, if already exists
    utils::FileDetails dirDetails = utils::GetFileDetails((this->root + "/index.bin").c_str());
 
    uint32 n = 0;
    utils::FileDetails altDetails;
    // Find backup location
    std::string altRoot = utils::StringFormat("%s.%u", this->root.c_str(), n);
 
    ReplayIndexEntry initEntry;
    initEntry.reset();
 
    // if recording
    if (dirDetails.doesExist) {
        while ((altDetails = utils::GetFileDetails((altRoot + "/index.bin").c_str())).doesExist)
            altRoot = utils::StringFormat("%s.%u", this->root.c_str(), ++n);
 
        if (!utils::MoveAFile(this->root.c_str(), altRoot.c_str(), true)) {
            LogPrint(0, 0, 0, "Cannot move the old storage location '%s' - using the alternate location '%s' instead", this->root.c_str(), altRoot.c_str());
            this->root = altRoot;
        }
        else
            LogPrint(0, 0, 0, "Old storage location '%s' backed up to '%s'", this->root.c_str(), altRoot.c_str());
    }
    indexFile = this->root + "/index.bin";
    metadataFile = this->root + "/metadata.bin";
    metadataTextFile = this->root + "/metadata.text";
    csvFile = this->root + "/data.csv";
 
    if (utils::GetFileDetails(this->root.c_str()).doesExist) {
        if (!utils::WriteAFile(indexFile.c_str(), (char*)&initEntry, sizeof(ReplayIndexEntry), true)) {
            LogPrint(0, 0, 0, "Cannot create index file in storage location '%s'", this->root.c_str());
            return false;
        }
        else
            LogPrint(0, 0, 0, "Using existing unused recording storage location '%s'", this->root.c_str());
    }
    else {
        if (!utils::CreateADir(this->root.c_str())) {
            LogPrint(0, 0, 0, "Cannot create storage location '%s'", this->root.c_str());
            return false;
        }
        else {
            if (!utils::WriteAFile(indexFile.c_str(), (char*)&initEntry, sizeof(ReplayIndexEntry), true)) {
                LogPrint(0, 0, 0, "Cannot create index file in storage location '%s'", this->root.c_str());
                return false;
            }
            else
                LogPrint(0, 0, 0, "Using new recording storage location '%s'", this->root.c_str());
        }
    }
 
    // create initial in-memory index
    if (maxCount)
        indexSize = maxCount * 2;
    else
        indexSize = 4096;
    indexData = (char*)malloc(indexSize * sizeof(ReplayIndexEntry));
    memset(indexData, 0, indexSize * sizeof(ReplayIndexEntry));
    return true;
}
 
 
 
bool MessagePlayer::addMessage(const char* name, DataMessage* msg) {
    if (!msg)
        return false;
    if (strlen(name) > MAXKEYNAMELEN) {
        LogPrint(0, 0, 0, "Trigger name '%s' too long to store", name);
        return false;
    }
 
    ReplayIndexEntry* entry = &((ReplayIndexEntry*)indexData)[curIndex];
    entry->reset();
    entry->serial = ++serial;
    entry->msgTime = lastReceiveTime = msg->getCreatedTime();
    entry->size = msg->getSize();
    entry->msg = new DataMessage(*msg);
    utils::strcpyavail(entry->msgTriggerName, name, 127, false);
    curIndex++;
    curSize += entry->size;
    curCount++;
    if (curIndex == indexSize)
        curIndex = 0;
    if (curCount > indexSize - 10) {
        // expand memory
        char* newIndexData = (char*)malloc(indexSize * 2 * sizeof(ReplayIndexEntry));
        if (curIndex > curStart)
            memcpy(newIndexData, indexData + (curStart * sizeof(ReplayIndexEntry)), curCount * sizeof(ReplayIndexEntry));
        else if (curIndex < curStart) {
            memcpy(newIndexData, indexData + (curStart * sizeof(ReplayIndexEntry)),
                (indexSize - curStart) * sizeof(ReplayIndexEntry));
            memcpy(newIndexData + ((indexSize - curStart) * sizeof(ReplayIndexEntry)),
                indexData, curIndex * sizeof(ReplayIndexEntry));
        }
        if (curWrite < curStart)
            curWrite += (indexSize - curStart);
        else
            curWrite -= curStart;
        curStart = 0;
        curIndex = curCount;
        free(indexData);
        indexData = newIndexData;
        indexSize *= 2;
        memset(indexData + (curCount * sizeof(ReplayIndexEntry)), 0, (indexSize - curCount) * sizeof(ReplayIndexEntry));
    }
 
    if (lastReceiveTime && (lastReceiveTime - lastWriteTime > 1000000))
        writeBuffer();
 
    //printf("Added message...\n");
    return true;
}
 
bool MessagePlayer::flushToDisk() {
    return writeBuffer();
}
 
 
DataMessage* MessagePlayer::waitForNextMessage(uint32 ms, const char* &triggerName) {
    uint32 msToNext = 0;
    return waitForNextMessage(ms, triggerName, msToNext);
}
 
DataMessage* MessagePlayer::waitForNextMessage(uint32 ms, uint32 &msToNext) {
    const char* triggerName;
    return waitForNextMessage(ms, triggerName, msToNext);
}
 
DataMessage* MessagePlayer::waitForNextMessage(uint32 ms, const char* &triggerName, uint32 &msToNext) {
 
    uint64 now = GetTimeNow();
    if (nextMsgTime) {
        uint64 nextReadTime = lastReadTime + (nextMsgTime * 1000);
        if (nextReadTime > now + (ms*1000)) {
            //printf("Pretend waiting for %ums\n", ms);
            utils::Sleep(ms);
            msToNext -= ms;
            return NULL;
        }
        else if (nextReadTime > now) {
            //printf("Waiting for %ums\n", (uint32)((nextReadTime - now)/1000));
            utils::Sleep((uint32)((nextReadTime - now) / 1000));
        }
        else {
            //printf("Not waiting\n");
        }
    }
    else {
        //printf("Going straight for it\n");
    }
 
    // plan to send out next message
    char* data;
    uint32 datasize;
    DataMessage *msg = NULL;
 
    // ############ read into buffer
 
    if ((curIndex == indexSize) && rotate) {
        LogPrint(0, 0, 1, "Finished sending %u messages, rotating back to start...", curIndex);
        curIndex = 0;
    }
    if (curIndex < indexSize) {
        ReplayIndexEntry* entry = &((ReplayIndexEntry*)indexData)[curIndex];
        if (entry->size) {
            if (!(data = utils::ReadAFile(utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str(), datasize, true)))
                return NULL;
            msg = new DataMessage(data, false);
            msg->setType(NOTYPE);
            msg->setContextChange(NOCONTEXT);
            msg->setCreatedTime(lastReadTime = now);
            // api->postOutputMessage(entry->msgTriggerName, msg);
            utils::strcpyavail(currentTriggerName, entry->msgTriggerName, MAXKEYNAMELEN, true);
            triggerName = currentTriggerName;
            if (interval == 1) {
                if (curIndex == indexSize - 1)
                    nextMsgTime = msToNext = 1000;
                else if (((ReplayIndexEntry*)indexData)[curIndex + 1].msgTime > entry->msgTime)
                    nextMsgTime = msToNext = (uint32)((((ReplayIndexEntry*)indexData)[curIndex + 1].msgTime - entry->msgTime) / 1000);
                else
                    nextMsgTime = msToNext = 0;
            }
        }
        else
            LogPrint(0, 0, 0, "Couldn't find message for index %u...", curIndex);
        curIndex++;
        return msg;
    }
    else if (curIndex == indexSize) {
        LogPrint(0, 0, 1, "Finished sending %u messages...", curIndex);
        curIndex++;
        nextMsgTime = msToNext = 0;
    }
    return NULL;
}
 
DataMessage* MessagePlayer::waitForNextMessage(uint32 ms) {
    const char* temp;
    return waitForNextMessage(ms, temp);
}
 
std::string MessagePlayer::getCurrentTriggerName() {
    return currentTriggerName;
}
 
 
bool MessagePlayer::writeBuffer() {
    // If we are a reader
    if (interval)
        return true;
 
    ReplayIndexEntry* entry;
    char* data;
    uint32 datasize;
    std::string csvOutput;
 
    // write unstored messages to files
    if (curWrite > curIndex) {
        // serial has looped, run to the end
        while (curWrite < indexSize) {
            entry = &((ReplayIndexEntry*)indexData)[curWrite];
            utils::WriteAFile(utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str(), (char*)entry->msg->data, entry->size, true);
            csvOutput += entry->msg->toCSV(",", utils::StringFormat("%llu,%s", entry->serial, entry->msgTriggerName).c_str(), &subtypeList, &subcontextList, &componentNameList);
            //printf("Wrote message '%s'...\n", utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str());
            delete(entry->msg);
            entry->msg = NULL;
            curWrite++;
        }
        curWrite = 0;
    }
    while (curWrite < curIndex) {
        entry = &((ReplayIndexEntry*)indexData)[curWrite];
        utils::WriteAFile(utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str(), (char*)entry->msg->data, entry->size, true);
        csvOutput += entry->msg->toCSV(",", utils::StringFormat("%llu,%s", entry->serial, entry->msgTriggerName).c_str(), &subtypeList, &subcontextList, &componentNameList);
        //printf("Wrote message '%s'...\n", utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str());
        delete(entry->msg);
        entry->msg = NULL;
        curWrite++;
    }
    while ((maxCount && (curCount > maxCount)) || (maxSize && (curSize > maxSize))) {
        entry = &((ReplayIndexEntry*)indexData)[curStart];
        // delete oldest file
        utils::DeleteAFile(utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str(), true);
        //printf("Delete message '%s'...\n", utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str());
        curSize -= entry->size;
        curCount--;
        curStart++;
        if (curStart == indexSize)
            curStart = 0;
        entry->size = 0;
    }
    lastWriteTime = GetTimeNow();
    // write index file
    if (curIndex > curStart) {
        if (!utils::WriteAFile(indexFile.c_str(), indexData + (curStart * sizeof(ReplayIndexEntry)), curCount * sizeof(ReplayIndexEntry), true))
            LogPrint(0, 0, 0, "Couldn't write index file '%s'", indexFile.c_str());
        //else
        //  api->logPrint(0, "Wrote index file '%s' size %llu [%u]", indexFile.c_str(), curCount * sizeof(ReplayIndexEntry), curCount);
    }
    else if (curIndex < curStart) {
        datasize = curCount * sizeof(ReplayIndexEntry);
        data = new char[datasize];
        memcpy(data, indexData + (curStart * sizeof(ReplayIndexEntry)),
            (indexSize - curStart) * sizeof(ReplayIndexEntry));
        memcpy(data + ((indexSize - curStart) * sizeof(ReplayIndexEntry)),
            indexData, curIndex * sizeof(ReplayIndexEntry));
        if (!utils::WriteAFile(indexFile.c_str(), data, datasize, true))
            LogPrint(0, 0, 0, "Couldn't write index file '%s'", indexFile.c_str());
        //else
        //  api->logPrint(0, "Wrote index temp file '%s' size %llu [%u]", indexFile.c_str(), datasize, curCount);
        delete[] data;
    }
    if (!utils::DoesAFileExist(csvFile.c_str()))
        csvOutput = utils::StringFormat("%s%s", DataMessage::GetCSVHeader(",", "serial,triggername").c_str(), csvOutput.c_str());
    utils::AppendToAFile(csvFile.c_str(), csvOutput.c_str(), (uint32)csvOutput.length(), false);
    //printf("Wrote messages...\n");
    return true;
}
 
bool MessagePlayer::readBuffer() {
    return false;
}
 
 
std::string MessagePlayer::printAllString(const char* format) {
 
    bool json = format ? (utils::stristr(format, "json") != NULL) : false;
    bool xml = format ? (utils::stristr(format, "xml") != NULL) : false;
    bool text = format ? (utils::stristr(format, "text") != NULL) : false;
 
    char* data;
    uint32 datasize;
    std::string output;
    ReplayIndexEntry* entry;
    DataMessage* msg;
 
    output += utils::StringFormat("Msg recording at %s contains %u messages\n\n", root.c_str(), curCount);
 
    for (uint32 i = 0; i < indexSize; i++) {
        entry = &((ReplayIndexEntry*)indexData)[i];
        if (entry->size) {
            output += utils::StringFormat("[%u]\t Msg at %s\t size %s\t trigger: '%s'\n", i+1,
                PrintTimeString(entry->msgTime).c_str(), utils::BytifySize(entry->size).c_str(), entry->msgTriggerName);
            if (format) {
                data = utils::ReadAFile(utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str(), datasize, true);
                if (data) {
                    msg = new DataMessage(data, false);
                    if (json)
                        output += msg->toJSON(&subtypeList, &subcontextList, &componentNameList) + "\n\n";
                    if (xml)
                        output += msg->toXML(&subtypeList, &subcontextList, &componentNameList) + "\n";
                    if (text)
                        output += msg->getUserEntriesAsString() + "\n";
                    delete msg;
                }
            }
        }
    }
 
    return output;
}
 
bool MessagePlayer::exportToCSVFile(const char* filename, const char* separator) {
    std::string csv = exportToCSV(separator);
    if (!csv.length())
        return false;
    return utils::WriteAFile(filename, csv.c_str(), (uint32)csv.length());
}
 
std::string MessagePlayer::exportToCSV(const char* separator) {
 
    char* data;
    uint32 datasize;
    std::string output;
    ReplayIndexEntry* entry;
    DataMessage* msg;
 
    output = DataMessage::GetCSVHeader(",", "serial,triggername");
 
    for (uint32 i = 0; i < indexSize; i++) {
        entry = &((ReplayIndexEntry*)indexData)[i];
        if (entry->size) {
            data = utils::ReadAFile(utils::StringFormat("%s/%llu.msg", root.c_str(), entry->serial).c_str(), datasize, true);
            if (data) {
                msg = new DataMessage(data, false);
                output += msg->toCSV(",", utils::StringFormat("%llu,%s", entry->serial, entry->msgTriggerName).c_str(), &subtypeList, &subcontextList, &componentNameList);
                delete msg;
            }
        }
    }
 
    return output;
}
 
 
bool MessagePlayer::setSubTypeList(std::map<uint16, std::string>& subtypes) {
    subtypeList.insert(subtypes.begin(), subtypes.end());
    return writeMetadata();
}
 
bool MessagePlayer::setSubContextList(std::map<uint16, std::string>& subcontexts) {
    subcontextList.insert(subcontexts.begin(), subcontexts.end());
    return writeMetadata();
}
 
bool MessagePlayer::setComponentNameList(std::map<uint32, std::string>& names) {
    componentNameList.insert(names.begin(), names.end());
    return writeMetadata();
}
 
bool MessagePlayer::writeMetadata() {
    // If we are a reader
    if (interval)
        return false;
 
    std::map<uint16, std::string>::iterator iType = subtypeList.begin(), eType = subtypeList.end();
    std::map<uint16, std::string>::iterator iContext = subcontextList.begin(), eContext = subcontextList.end();
    std::map<uint32, std::string>::iterator iComp = componentNameList.begin(), eComp = componentNameList.end();
 
    std::stringstream metadataStream;
    DataMessage* metadata = new DataMessage();
    while (iType != eType) {
        metadata->setString(iType->first, "subtype", iType->second.c_str());
        metadataStream << "Subtype: " << iType->first << "=" << iType->second << "\n";
        iType++;
    }
 
    while (iContext != eContext) {
        metadata->setString(iContext->first, "subcontext", iContext->second.c_str());
        metadataStream << "Subcontext: " << iContext->first << "=" << iContext->second << "\n";
        iContext++;
    }
 
    while (iComp != eComp) {
        metadata->setString(iComp->first, "component", iComp->second.c_str());
        metadataStream << "Component: " << iComp->first << "=" << iComp->second << "\n";
        iComp++;
    }
 
    if (!utils::WriteAFile(metadataFile.c_str(), (char*)(metadata->data), metadata->data->size, true))
        return false;
    if (!utils::WriteAFile(metadataTextFile.c_str(), metadataStream.str().c_str(), (uint32)metadataStream.str().length(), false))
        return false;
 
    return true;
}
 
bool MessagePlayer::setSystemIDs(DataMessage* msg) {
    if (!msg || !msg->data->userCount)
        return false;
 
    std::map<int64, std::string> stringArray;
    std::map<int64, std::string>::iterator i, e;
 
    stringArray = msg->getStringArray("subtype");
    if (stringArray.size()) {
        subtypeList.clear();
        i = stringArray.begin();
        e = stringArray.end();
        while (i != e) {
            subtypeList[(uint16)(i->first)] = i->second;
            i++;
        }
    }
 
    stringArray = msg->getStringArray("subcontext");
    if (stringArray.size()) {
        subcontextList.clear();
        i = stringArray.begin();
        e = stringArray.end();
        while (i != e) {
            subcontextList[(uint16)(i->first)] = i->second;
            i++;
        }
    }
 
    stringArray = msg->getStringArray("component");
    if (stringArray.size()) {
        componentNameList.clear();
        i = stringArray.begin();
        e = stringArray.end();
        while (i != e) {
            componentNameList[(uint32)(i->first)] = i->second;
            i++;
        }
    }
    writeMetadata();
    return true;
}
 
bool MessagePlayer::readMetadata() {
    // If we are a writer
    if (!interval)
        return false;
 
    uint32 size;
    char* data = utils::ReadAFile(metadataFile.c_str(), size, true);
    if (!data || !size)
        return false;
 
    DataMessage* metadata = new DataMessage(data, false);
 
    bool result = setSystemIDs(metadata);
    delete metadata;
    return result;
}
 
 
 
} // namespace cmlabs