CommunicationRecorder.cpp

Go to the documentation of this file.

/***************************** License **********************************

 Copyright (C) 2008 by Communicative Machines
 http://www.cmlabs.com   All rights reserved.

 This library is free software; you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public
 License as published by the Free Software Foundation; either
 version 2.1 of the License, or (at your option) any later version.

 This library is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 Lesser General Public License for more details.

 You should have received a copy of the GNU Lesser General Public
 License along with this library; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

************************************************************************/
// NetOriginalMessageProtocol.cpp: implementation of the NetOriginalMessageProtocol class.
//
//////////////////////////////////////////////////////////////////////

#include "CommunicationRecorder.h"

namespace cmlabs {

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

CommunicationRecorder::CommunicationRecorder(const JString& storage, bool overwrite) : JThread() {
        resetFlag = false;
        overwriteRecording = overwrite;
        shouldContinue = true;
        accessType = COMRECNOINIT;
        if ( (storage.endsWith("/")) || (storage.endsWith("\\")) )
                this->storage = storage.substring(0, storage.length()-2);
        else
                this->storage = storage;
        indexFilename = "_index.xml";
        lastMsgIndex = -1;
        lastSampleIndex = -1;
        masterOffset = 0;
        paused = false;
}

CommunicationRecorder::~CommunicationRecorder() {
        finishWriting(3000);
        shouldContinue = false;
        wait(50);
        terminateWait(50);
}

Object* CommunicationRecorder::clone() const {
        return NULL;
}

bool CommunicationRecorder::initReading() {
        accessType = COMRECREADER;
        // Initialise existing storage location
        // First check if it exists
        if (!JFile::doesADirExist(storage))
                return false;
        // Initialise database
        if (!loadIndexFile())
                return false;
        return true;
}

InfoItem* CommunicationRecorder::getRecordingInfo(const JString& storage) {
        if (!JFile::doesADirExist(storage))
                return false;
        InfoItem* info = getIndexFileInfo(storage, "_index.xml");
        if (info == NULL)
                return NULL;
        Collection files = JFile::getFilesInADir(storage);
        int msgFiles = 0, sampleFiles = 0, binFiles = 0;
        for (JString file = files.getFirst(); file.length() > 0; file = files.getNext()) {
                if (file.endsWith(".msg"))
                        msgFiles++;
                else if (file.endsWith(".sample"))
                        sampleFiles++;
                else if (file.endsWith(".bin"))
                        binFiles++;
        }
        // Get number of Message stored
        info->setEntry("Number of Messages", JString(msgFiles));
        // Get number of Samples stored
        info->setEntry("Number of DataSamples", JString(sampleFiles));
        // Get number of Binary Attachments stored
        info->setEntry("Number of Binary Attachments", JString(binFiles));
        return info;
}

bool CommunicationRecorder::initWriting() {
        accessType = COMRECWRITER;
        // Initialise new storage location
        // First check if it exists already
        int n = 1;
        if (JFile::doesADirExist(storage)) {
                // If so, should we overwrite?
                if (overwriteRecording) {
                        if (!JFile::deleteADir(storage, true))
                                return false;
                }
                else {
                        // If so, move out of the way
                        JString moveTo = JString::format("%s.bak", (char*) storage);
                        while (JFile::doesADirExist(moveTo))
                                moveTo = JString::format("%s.bak%d", (char*) storage, n++);
                        if (!JFile::moveADir(storage, moveTo))
                                return false;
                }
        }
        // Right, we are now ready to create a fresh database
        if (!JFile::createADir(storage, true))
                return false;
        if (!saveIndexFile())
                return false;

        // Initialise fresh database
        return true;
}

void CommunicationRecorder::run() {

        JTime lastMessageWarning;
        JTime lastSampleWarning;
        Message* msg;
        DataSample* sample;
        JString basefilename;
        while (shouldContinue) {
                if ( (msg = (Message*) msgQueue.waitForNewEntry(20)) != NULL ) {
                        // Find next index number
                        lastMsgIndex++;
                        // Generate filename
                        basefilename = JString::format("%s/Message%.6d", (char*) storage, lastMsgIndex);
                        if (!saveMessage(basefilename, msg)) {
                                printf("CommunicationRecorder: Error saving '%s' to '%d'\n", 
                                        (char*) basefilename, (char*) storage);
                                delete(msg);
                        }
                }
                if ( (sample = (DataSample*) sampleQueue.waitForNewEntry(20)) != NULL ) {
                        // Find next index number
                        lastSampleIndex++;
                        // Generate filename
                        basefilename = JString::format("%s/Sample%.6d", (char*) storage, lastSampleIndex);
                        if (!saveSample(basefilename, sample)) {
                                printf("CommunicationRecorder: Error saving '%s' to '%d'\n", 
                                        (char*) basefilename, (char*) storage);
                                delete(sample);
                        }
                }
                if ((msgQueue.getCount() > 100) && (lastMessageWarning.getAge() > 3000)) {
                        printf("CommunicationRecorder: Message Queue Warning - %d messages waiting to be written to disk\n", msgQueue.getCount());
                        lastMessageWarning.reset();
                }
                if ((sampleQueue.getCount() > 100) && (lastSampleWarning.getAge() > 3000)) {
                        printf("CommunicationRecorder: Stream Queue Warning - %d samples waiting to be written to disk\n", sampleQueue.getCount());
                        lastSampleWarning.reset();
                }
        }
}

bool CommunicationRecorder::wasReset() {
        if (!resetFlag)
                return false;
        resetFlag = false;
        return true;
}

bool CommunicationRecorder::isPaused() {
        return paused;
}

bool CommunicationRecorder::isInitialised() {
        return (accessType != COMRECNOINIT);
}

bool CommunicationRecorder::isWriter() {
        return (accessType == COMRECWRITER);
}

bool CommunicationRecorder::isReader() {
        return (accessType == COMRECREADER);
}


// Operation
bool CommunicationRecorder::setNewStorageLocation(const JString& storage) {
        if (storage.length() == 0)
                return false;
        if ( (storage.endsWith("/")) || (storage.endsWith("\\")) )
                this->storage = storage.substring(0, storage.length()-2);
        else
                this->storage = storage;
        indexFilename = "_index.xml";
        lastMsgIndex = -1;
        lastSampleIndex = -1;
        masterOffset = 0;
        resetFlag = true;
        return true;
}

bool CommunicationRecorder::startRecording() {
        paused = false;
        return true;
}

bool CommunicationRecorder::pauseRecording() {
        paused = true;
        return true;
}

bool CommunicationRecorder::restartRecording() {
        paused = false;
        lastMsgIndex = -1;
        lastSampleIndex = -1;
        resetFlag = true;
        return initWriting();
}

bool CommunicationRecorder::startPlayback() {
        paused = false;
        return true;
}

bool CommunicationRecorder::pausePlayback() {
        paused = true;
        return true;
}

bool CommunicationRecorder::restartPlayback() {
        paused = false;
        resetFlag = true;
        lastMsgIndex = -1;
        lastSampleIndex = -1;
        return initReading();
}


// General Queries
long CommunicationRecorder::getCurrentOffset() {
        return currentStartTime.getAge();
}

JTime CommunicationRecorder::getStartTime() {
        return currentStartTime;
}

JTime CommunicationRecorder::calcOriginalTime(const JTime& time) {
        return time - masterOffset;
}

JTime CommunicationRecorder::calcCurrentTime(const JTime& originalTime) {
        return originalTime + masterOffset;
}

int CommunicationRecorder::getMessageCount() {
        return lastMsgIndex;
}

int CommunicationRecorder::getSampleCount() {
        return lastSampleIndex;
}

JTime CommunicationRecorder::getFirstMessageTime() {
        // ##################################### //
        return JTime::createInvalid();
}

JTime CommunicationRecorder::getLastMessageTime() {
        // ##################################### //
        return JTime::createInvalid();
}

JTime CommunicationRecorder::getFirstSampleTime() {
        // ##################################### //
        return JTime::createInvalid();
}

JTime CommunicationRecorder::getLastSampleTime() {
        // ##################################### //
        return JTime::createInvalid();
}

long CommunicationRecorder::getFirstMessageOffset() {
        // ##################################### //
        return -1;
}

long CommunicationRecorder::getLastMessageOffset() {
        // ##################################### //
        return -1;
}

long CommunicationRecorder::getFirstSampleOffset() {
        // ##################################### //
        return -1;
}

long CommunicationRecorder::getLastSampleOffset() {
        // ##################################### //
        return -1;
}

// Adding Data
bool CommunicationRecorder::addNewMessage(Message* msg) {
        if (msg == NULL)
                return false;
        msgQueue.add(msg);
        return true;
}

bool CommunicationRecorder::addNewSample(DataSample* sample) {
        if (sample == NULL)
                return false;
        sampleQueue.add(sample);
        return true;
}

// Retrieving Data
Message* CommunicationRecorder::retrieveMessage(int n) {
        return loadMessage(JString::format("%s/Message%.6d", (char*) storage, n));
}

DataSample* CommunicationRecorder::retrieveSample(int n) {
        return loadSample(JString::format("%s/Sample%.6d", (char*) storage, n));
}

Message* CommunicationRecorder::retrieveNextMessage(const JTime& time) {
        // ##################################### //
        return NULL;
}

DataSample* CommunicationRecorder::retrieveNextSample(const JTime& time) {
        // ##################################### //
        return NULL;
}









bool CommunicationRecorder::saveSegment(int start, int end) {
        // ##################################### //
        return false;
}
bool CommunicationRecorder::loadSegment(int start, int end) {
        // ##################################### //
        return false;
}



Message* CommunicationRecorder::loadMessage(const JString& basefilename) {

        JString xml = JFile::readAFileASCII(JString::format("%s.msg", (char*) basefilename));
        if (xml.length() == 0)
                return NULL;

        Message* msg = new Message();
        if (!msg->fromXML(xml)) {
                delete(msg);
                return NULL;
        }

        // Now check for binary attachments
        char* buffer;
        int size;
        int i = 0;
        
        buffer = JFile::readAFileBinary(JString::format("%s.%d.bin", (char*) basefilename, i), size);
        while ( (buffer != NULL) && (size > 0)) {
                msg->fromBinaryBuffer(i, buffer, size);
                delete [] buffer;
                i++;
                buffer = JFile::readAFileBinary(JString::format("%s.%d.bin", (char*) basefilename, i), size);
        }

        // Adjust timestamps
        if (msg->postedTime.isValid())
                msg->postedTime += masterOffset;
        if (msg->receivedTime.isValid())
                msg->receivedTime += masterOffset;

        return msg;
}

DataSample* CommunicationRecorder::loadSample(const JString& basefilename) {
        
        int size;
        char* buffer = JFile::readAFileBinary(JString::format("%s.sample", (char*) basefilename), size);

        if ( (buffer == NULL) || (size <= 0)) {
                delete [] buffer;
                return NULL;
        }
        
        DataSample* sample = new DataSample();
        if (!sample->fromBinaryBuffer(0, buffer, size)) {
                delete [] buffer;
                delete(sample);
                return NULL;
        }
        delete [] buffer;

        // Adjust timestamps
        if (sample->timestamp.isValid())
                sample->timestamp += masterOffset;

        return sample;
}


bool CommunicationRecorder::saveMessage(const JString& basefilename, Message* msg) {
        if (msg == NULL) return false;

        char* buffer;
        int binChunkCount = msg->getBinaryChunkCount();
        int maxBinSize = 0;
        JString filename;
        
        if (binChunkCount > 0) {
                long* bufSizes = new long[binChunkCount];
                long subBinSize;
                int i;
                for (i=0; i<binChunkCount; i++) {
                        subBinSize = msg->getBinarySize(i);
                        bufSizes[i] = subBinSize;
                        if (subBinSize > maxBinSize)
                                maxBinSize = subBinSize;
                }
                buffer = new char[maxBinSize];
                for (i=0; i<binChunkCount; i++) {
                        subBinSize = bufSizes[i];
                        msg->toBinaryBuffer(i, buffer, maxBinSize);
                        if (!JFile::writeAFileBinary(JString::format("%s.%d.bin", (char*) basefilename, i), buffer, subBinSize)) {
                                delete [] bufSizes;
                                delete [] buffer;
                                return false;
                        }
                }
                delete [] bufSizes;
                delete [] buffer;
        }

        JString xml = msg->toXML();
        if (!JFile::writeAFileASCII(JString::format("%s.msg", (char*) basefilename), xml))
                return false;

        delete(msg);
        return true;
}

bool CommunicationRecorder::saveSample(const JString& basefilename, DataSample* sample) {

        if (sample == NULL) return false;

        long size = sample->getBinarySize(0);
        char* buffer = new char[size];
        sample->toBinaryBuffer(0, buffer, size);
        if (!JFile::writeAFileBinary(JString::format("%s.sample", (char*) basefilename), buffer, size)) {
                delete [] buffer;
                return false;
        }

        delete [] buffer;
        delete(sample);
        return true;
}

bool CommunicationRecorder::finishWriting(int timeout) {
        JTime start;
        bool result = false;
        while (start.getAge() < timeout) {
                if ( (msgQueue.getCount() == 0) && (sampleQueue.getCount() == 0) ) {
                        result = true;          
                        break;
                }
                wait(10);
        }
        if (!result)
                printf("FinishWriting: Still %d messages and %d samples waiting...\n",
                        msgQueue.getCount(), sampleQueue.getCount());
        return result;
}

bool CommunicationRecorder::saveIndexFile() {
        if (!isWriter()) return false;
        JString xml = JString::format("<!-- Original Start Time: %s -->\n", (char*) originalStartTime.print());
        xml += originalStartTime.toXML("originaltime");
        JString data = JString::format("<CommunicationRecorder>\n%s</CommunicationRecorder>\n", (char*) xml.indentXML());
        JString filename = JString::format("%s/%s", (char*) storage, (char*) indexFilename);
        return JFile::writeAFileASCII(filename, data);
}

InfoItem* CommunicationRecorder::getIndexFileInfo(const JString& location, const JString& indexfile) {
        JString filename = JString::format("%s/%s", (char*) location, (char*) indexfile);
        JString xml = JFile::readAFileASCII(filename);
        if (xml.length() == 0)
                return NULL;
        XMLParser parser;
        if (!parser.parseXML(xml))
                return NULL;
        XMLNode* rootNode = parser.getRootNode();
        if (rootNode == NULL)
                return NULL;
        if (!rootNode->getTag().equalsIgnoreCase("CommunicationRecorder"))
                return NULL;
        XMLNode* node = rootNode->getChildNode("originaltime");
        if (node == NULL)
                return NULL;
        JTime oriTime;
        if (!oriTime.fromXML(node))
                return NULL;
        InfoItem* info = new InfoItem();
        info->setTime("Original Start Time", oriTime);
        info->setEntry("Storage Location", location);
        info->setEntry("Index File", indexfile);
        return info;
}

bool CommunicationRecorder::loadIndexFile() {
        if (!isReader()) return false;
        JString filename = JString::format("%s/%s", (char*) storage, (char*) indexFilename);
        JString xml = JFile::readAFileASCII(filename);
        if (xml.length() == 0)
                return false;
        XMLParser parser;
        if (!parser.parseXML(xml))
                return false;
        XMLNode* rootNode = parser.getRootNode();
        if (rootNode == NULL)
                return false;
        if (!rootNode->getTag().equalsIgnoreCase("CommunicationRecorder"))
                return false;
        XMLNode* node = rootNode->getChildNode("originaltime");
        if (node == NULL)
                return false;
        if (!originalStartTime.fromXML(node))
                return false;
        masterOffset = currentStartTime - originalStartTime;
        return true;
}


bool CommunicationRecorder::unitTest() {

        // This is the running Communication Writer, so first write some messages and samples
        Message* msg;
        DataSample* sample;
        char* data;
        int n;

//      addUnitTestLog(JString::format("Multiple writers with flag and time difference failed, only received %d of %d...", c, count*2));
        addUnitTestLog("Adding Messages and Samples to the Recorder...");

        JTime stamp;
        for (n=0; n<10; n++) {
                msg = new Message("Me", "You", "MyType", JString("Message: ") + JString(n), "");
                msg->postedTime = stamp;
                this->addNewMessage(msg);
                sample = new DataSample("Raw", JString("Sample: ") + JString(n));
                // ------------------
                stamp += 10;
                data = new char[1024];
                memset(data, n, 1024);
                sample->giveData(data, 1024);
                sample->timestamp = stamp;
                this->addNewSample(sample);
                stamp += 10;
        }

        wait(200);

        if (!finishWriting(1000)) {
                addUnitTestLog("Recorder finishWriting failed...");
                return false;
        }

        CommunicationRecorder* comReader = new CommunicationRecorder(this->storage);
        if (!comReader->initReading()) {
                delete(comReader);
                addUnitTestLog("Initialising Reader failed...");
                return false;
        }
        comReader->start();

        // Check that we can read the above written messages and samples back by position
        for (n=0; n<10; n++) {
                msg = comReader->retrieveMessage(n);
                if (msg == NULL) {
                        addUnitTestLog(JString::format("Reading message %d failed...", n));
                        delete(comReader);
                        return false;
                }
                else if (!msg->content.equals(JString("Message: ") + JString(n))) {
                        addUnitTestLog(JString::format("Reading message %d was wrong: '%s'...", n, (char*) msg->content));
                        delete(msg);
                        delete(comReader);
                        return false;
                }
                delete(msg);

                sample = comReader->retrieveSample(n);
                if (sample == NULL) {
                        addUnitTestLog(JString::format("Reading sample %d failed...", n));
                        delete(comReader);
                        return false;
                }
                else if (!sample->name.equals(JString("Sample: ") + JString(n))) {
                        addUnitTestLog(JString::format("Reading sample %d was wrong: '%s'...", n, (char*) sample->name));
                        delete(sample);
                        delete(comReader);
                        return false;
                }
                delete(sample);
        }
        
        // Check that we can read the above written messages and samples back by time
/*      stamp = comReader->getStartTime();
        for (n=0; n<10; n++) {
                msg = comReader->retrieveNextMessage(stamp);
                if (msg == NULL) {
                        addUnitTestLog(JString::format("Reading message %d failed...", n));
                        delete(comReader);
                        return false;
                }
                else if (!msg->content.equals(JString("Message: ") + JString(n))) {
                        addUnitTestLog(JString::format("Reading message %d was wrong: '%s'...", n, (char*) msg->content));
                        delete(msg);
                        delete(comReader);
                        return false;
                }
                delete(msg);

                sample = comReader->retrieveNextSample(stamp);
                if (sample == NULL) {
                        addUnitTestLog(JString::format("Reading sample %d failed...", n));
                        delete(comReader);
                        return false;
                }
                else if (!sample->name.equals(JString("Sample: ") + JString(n))) {
                        addUnitTestLog(JString::format("Reading sample %d was wrong: '%s'...", n, (char*) sample->name));
                        delete(sample);
                        delete(comReader);
                        return false;
                }
                delete(sample);
        }
*/
        delete(comReader);
        return true;    
}

} // namespace cmlabs

---