C:/Active/libraries/cxutils/2.0/src/cxutils/ipc/mappedmessagebox.cpp

Go to the documentation of this file.

#include "cxutils/ipc/mappedmessagebox.h"
#include "cxutils/time.h"
#include <stdio.h>
#include <string.h>
#include <assert.h>
#include <limits.h>
#include <iostream>
#include <sstream>

using namespace std;
using namespace CxUtils;

MappedMessageBox::Header::Header() : mWriteTimeMs(0),
                                     mReadTimeMs(0),
                                     mCount(0),
                                     mStartBytePos(Header::Size),
                                     mEndBytePos(Header::Size)
{

}


MappedMessageBox::Header::Header(const MappedMessageBox::Header& header) :  mWriteTimeMs(0),
                                                                            mReadTimeMs(0),
                                                                            mCount(0),
                                                                            mStartBytePos(Header::Size),
                                                                            mEndBytePos(Header::Size)
{
    *this = header;
}


MappedMessageBox::Header::~Header()
{
}


MappedMessageBox::Header& MappedMessageBox::Header::operator =(const MappedMessageBox::Header& header)
{
    if(this != &header)
    {
        mWriteTimeMs = header.mWriteTimeMs;
        mReadTimeMs = header.mReadTimeMs;
        mCount = header.mCount;
        mStartBytePos = header.mStartBytePos;
        mEndBytePos = header.mEndBytePos;
    }
    return *this;
}


MappedMessageBox::MappedMessageBox() : mMessagesRead(0)
{
}


MappedMessageBox::~MappedMessageBox()
{
    CloseMessageBox();
}


bool MappedMessageBox::OpenMessageBox(const ID id)
{
    bool result = false;

    CloseMessageBox();

    if(mMessageBox.OpenMappedMemory(id) && mMessageBox.Length() > Header::Size)
    {
        result = true;
    }

    if(result == false)
    {
        CloseMessageBox();
    }

    return result;
}

bool MappedMessageBox::OpenMessageBox(const std::string& name)
{
    bool result = false;

    CloseMessageBox();

    if(mMessageBox.OpenMappedMemory(name) && mMessageBox.Length() > Header::Size)
    {
        result = true;
    }

    if(result == false)
    {
        CloseMessageBox();
    }

    return result;
}


bool MappedMessageBox::CreateMessageBox(const ID id,
                                        const unsigned int size)
{
    std::ostringstream input;
    input << id;
    return CreateMessageBox(input.str(), size);
}


bool MappedMessageBox::CreateMessageBox(const std::string& name,
                                        const unsigned int size)
{
    bool result = false;
    Header header;

    CloseMessageBox();
    
    if(mMessageBox.CreateMappedMemory(name, size + Header::Size))
    {
        result = true;        
        mMessageBox.Lock();
        header.mStartBytePos = Header::Size;
        header.mEndBytePos = Header::Size;
        header.mWriteTimeMs = GetTimeMs();
        header.mReadTimeMs = GetTimeMs();
        WriteHeader(header);
        mMessageBox.Unlock();
    }
    else if(mMessageBox.OpenMappedMemory(name))
    {
        // Check to see how active the
        // message box is, and if it is not
        // active, flush the contents
        mMessageBox.Lock();
        header = ReadHeader();
        if(mMessageBox.Length() < size + Header::Size)
        {
            if(GetTimeMs() - header.mWriteTimeMs > DefaultTimeOut && 
               GetTimeMs() - header.mReadTimeMs > DefaultTimeOut)
            {
                mMessageBox.Unlock();
                mMessageBox.CloseMappedMemory();
                MappedMemory::DeleteMappedMemory(name);
                CxUtils::SleepMs(DefaultTimeOut);
                // Try to create again.
                if(mMessageBox.CreateMappedMemory(name, size + Header::Size))
                {
                    result = true;        
                    mMessageBox.Lock();
                    header.mStartBytePos = Header::Size;
                    header.mEndBytePos = Header::Size;
                    header.mWriteTimeMs = GetTimeMs();
                    header.mReadTimeMs = GetTimeMs();
                    WriteHeader(header);
                    mMessageBox.Unlock();
                }
            }
        }
        else
        {
            Time::Stamp currentTimeMs = GetTimeMs();
#ifndef WIN32 // LINUX
            // This if statement is needed because sometimes in Linux shared memory is not closed/released
            // when a process exits, so we need to be prepared for memory used later in the previous day
            if(header.mWriteTimeMs > currentTimeMs && header.mWriteTimeMs - currentTimeMs >= DefaultTimeOut)
            {
                result = false;
            }
#endif
            if(currentTimeMs > header.mWriteTimeMs && currentTimeMs - header.mWriteTimeMs >= DefaultTimeOut)
            {
                result = true;
            } 
        }
        if(result)
        {
            header.mWriteTimeMs = GetTimeMs();
            WriteHeader(header);
        }
        mMessageBox.Unlock();
    }
    

    if(result == false)
    {
        CloseMessageBox();
    }

    return result;
}


void MappedMessageBox::CloseMessageBox()
{
    mMessageBox.CloseMappedMemory();
    mMessagesRead = 0;
}


bool MappedMessageBox::IsOpen() const
{
    if(mMessageBox.IsOpen())
    {
        return true;
    }
    return false;
}


bool MappedMessageBox::IsEmpty() const
{
    if(mMessageBox.IsOpen())
    {
        Header header;
        mMessageBox.Lock();
        header = ReadHeader();
        mMessageBox.Unlock();

        if(header.mCount == 0)
        {
            return true;
        }
    }

    return false;
}


bool MappedMessageBox::IsActive(const bool readTimeFlag, const unsigned int thresholdMilliseconds) const
{
    if(mMessageBox.IsOpen())
    {
        Header header;
        Time::Stamp timeMs = 0;
        mMessageBox.Lock();
        header = ReadHeader();
        mMessageBox.Unlock();
        Time::Stamp currentTimeMs = GetTimeMs();
        timeMs = readTimeFlag ? header.mReadTimeMs : header.mWriteTimeMs;
#ifndef WIN32 // LINUX
        // This if statement is needed because sometimes in Linux shared memory is not closed/released
        // when a process exits, so we need to be prepared for memory used later in the previous day
        if(timeMs >= currentTimeMs && timeMs - currentTimeMs >= thresholdMilliseconds)
        {
            return false;
        }
#endif
        if(currentTimeMs >= timeMs && currentTimeMs - timeMs <= thresholdMilliseconds)
        {
            return true;
        }
    }

    return false;
}


bool MappedMessageBox::Touch()
{
    if(mMessageBox.IsOpen())
    {
        Header header;
        mMessageBox.Lock();
        header = ReadHeader();
        header.mWriteTimeMs = GetTimeMs();
        WriteHeader(header);
        mMessageBox.Unlock();

        return true;
    }

    return false;
}


bool MappedMessageBox::WriteMessage(const Packet& message)
{
    bool result = false;
    if(mMessageBox.IsOpen())
    {
        Header header;
        mMessageBox.Lock();
        header = ReadHeader();
        // See if there is room in shared memory.
        unsigned int availableBytes = mMessageBox.Length() - Header::Size;
        unsigned int messageSize = message.Length() + MessageHeaderSize;
        if(messageSize < availableBytes)
        {
            // OK, so we know the message can fit in this box, but first
            // lets see if we need to delete old data.
            if(messageSize + header.mEndBytePos >= mMessageBox.Length() && header.mStartBytePos > Header::Size)
            {
                unsigned int dataSize = header.mEndBytePos - header.mStartBytePos;
                memmove(mMessageBox->Ptr() + Header::Size, 
                        mMessageBox->Ptr() + header.mStartBytePos, 
                        dataSize);
                header.mEndBytePos = dataSize + Header::Size;
                header.mStartBytePos = Header::Size;
            }

            if(messageSize + header.mEndBytePos < mMessageBox.Length() - 1)
            {
                mMessageBox->SetWritePos(header.mEndBytePos);
                if(mMessageBox->Write(((unsigned int)(message.Length()))) &&
                   mMessageBox->Write(message))
                {
                    result = true;
                    header.mEndBytePos += message.Length() + MessageHeaderSize;
                    header.mCount++;
                }
            }
        }
        header.mWriteTimeMs = GetTimeMs();
        WriteHeader(header);
        mMessageBox.Unlock();
    }
    
    return result;
}


bool MappedMessageBox::WriteMessage(const std::string& message)
{
    Packet::Wrapper wrapper((unsigned char *)message.c_str(), (unsigned int)message.size());
    return WriteMessage( *wrapper.GetData() );
}



bool MappedMessageBox::ReadMessage(Packet& message) const
{
    bool result = false;
    message.Clear();
    if(mMessageBox.IsOpen())
    {
        Header header;
        if(mMessageBox.Lock() == 0)
        {
            return result;
        }
        header = ReadHeader();
        
        if(header.mCount > 0 && header.mStartBytePos < header.mEndBytePos)
        {
            // Advance to the position of the next message we want to read.
            unsigned int messageSize = 0;
            unsigned int messagePosition = header.mStartBytePos;
            
            // Read the message size in the box.
            mMessageBox->Read(messageSize, messagePosition); 
            
            if(messageSize + MessageHeaderSize + messagePosition <= header.mEndBytePos)
            {
                if(mMessageBox->Read(message, messageSize, messagePosition + MessageHeaderSize) == (int)messageSize)
                {
                     (*( (unsigned *)(&mMessagesRead) ))++;
                    // Advance the start byte position in the array.
                    header.mStartBytePos = messagePosition + MessageHeaderSize + messageSize;
                    // Decrease the number of messages in the box.
                    header.mCount--;
                    result = true;
                }
            }
            if(messageSize + MessageHeaderSize > mMessageBox->Length())
            {
                assert("Corrupted Shared Memory Message Box" && 0);
            }
        }
        // Reset position values.
        if(header.mCount == 0)
        {
            header.mStartBytePos = header.mEndBytePos = Header::Size;
        }
        // Update the read time (shows we are actively checking for messages).
        header.mReadTimeMs = GetTimeMs();
        // Save updated header data.
        WriteHeader(header);
        // Release critical section.
        mMessageBox.Unlock();
    }
        
    return result;
}


bool MappedMessageBox::ReadMessage(std::string& message) const
{
    Packet pmessage;
    message.clear();
    if(ReadMessage(pmessage))
    {
        for(unsigned int i = 0; i < pmessage.Length(); i++)
        {
            message.push_back(pmessage.Ptr()[i]);
        }
        return true;
    }
    return false;
}


unsigned int MappedMessageBox::GetBufferSize() const
{
    if(mMessageBox.IsOpen())
    {
        return mMessageBox.Length() - Header::Size;
    }
    return 0;
}


unsigned int MappedMessageBox::GetNumMessages() const
{
    if(mMessageBox.IsOpen())
    {
        Header header;
        mMessageBox.Lock();
        header = ReadHeader();
        mMessageBox.Unlock();
        return header.mCount;
    }
    return 0;
}


MappedMessageBox::Header MappedMessageBox::ReadHeader() const
{
    Header header;
    if(mMessageBox.GetData() == NULL)
    {
        return header;
    }
    mMessageBox->SetReadPos(0);
    if(mMessageBox->Length() < Header::Size)
    {
        return header;
    }
    mMessageBox->Read(header.mWriteTimeMs);
    mMessageBox->Read(header.mReadTimeMs);
    mMessageBox->Read(header.mCount);
    mMessageBox->Read(header.mStartBytePos);
    mMessageBox->Read(header.mEndBytePos);
    return header;
}


void MappedMessageBox::WriteHeader(const MappedMessageBox::Header& header) const
{
    MappedMemory* memory = ((MappedMemory*)(&mMessageBox));
    if(memory->IsOpen() == false)
    {
        return;
    }
    (*memory)->SetWritePos(0);
    unsigned int totalWritten = 0;
    totalWritten += (*memory)->Write(header.mWriteTimeMs);
    totalWritten += (*memory)->Write(header.mReadTimeMs);
    totalWritten += (*memory)->Write(header.mCount);
    totalWritten += (*memory)->Write(header.mStartBytePos);
    totalWritten += (*memory)->Write(header.mEndBytePos);
}


/*  End of File */