jaus++/2.0/src/jaus/core/transport/transport.cpp

Go to the documentation of this file.

#include "jaus/core/transport/transport.h"
#include "jaus/core/events/event.h"
#include <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <iomanip>
#include <cxutils/fileio.h>

//  Include pthread library for LINUX
#ifdef WIN32

#else
#include <sys/sem.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#endif

using namespace JAUS;

const std::string Transport::Name = "urn:jaus:jss:core:Transport";

Transport::Receipt::Receipt()
{
    mPendingFlag = true;
    mpMessage = NULL;
    mSendTimeMs = mUpdateTimeMs = 0;
    mpResponses = NULL;
}


Transport::Receipt::~Receipt()
{
}


Transport::Transport() : Service(Service::ID(Transport::Name, 1.0), Service::ID())
{
    mLoggingFlag = false;
    mStopMessageProcessingFlag = false;
    mMessageProcessingDelayMs = 1;
    mNumberOfMessageProcessingThreads = 1;
    mJausMessageQueue.Reserve(MaxQueueSize);
    mSequenceNumber = 0;
}


Transport::~Transport()
{
    mStopMessageProcessingFlag = true;
    CxUtils::SleepMs(50);
    mMessageProcessingThreads.StopAllThreads();
    mPriorityJausMessagesThread.StopThread();
    std::map<UShort, Message*>::iterator m;
    for(m = mMessageTemplates.begin(); m != mMessageTemplates.end(); m++)
    {
        delete m->second;
    }
    mMessageTemplates.clear();
}


void Transport::SetMaxMessageProcessingThreads(const unsigned int maxThreads)
{
    if(IsInitialized() == false && maxThreads >= 1)
    {
        mNumberOfMessageProcessingThreads = maxThreads;
    }
}


void Transport::AddPriorityMessage(const UShort messageCode)
{
    if(IsInitialized() == false)
    {
        mPriorityJausMessageTypes.insert(messageCode);
        if(mPriorityJausMessagesThread.IsThreadActive() == false)
        {
            mPriorityJausMessagesThread.CreateThread(PriorityMessageProcessingThread, this);
            char buffer[512];
            sprintf(buffer, "%s_PRIORITY_THREAD_%d", mComponentID.ToString().c_str(), mMessageProcessingThreads.GetNumThreads() + 1);
        }
    }
}


void Transport::SetMessagePollingDelayMs(const unsigned int delayTimeMs)
{
    mMessageProcessingDelayMs = delayTimeMs;
}


Message* Transport::CreateMessage(const UShort messageCode) const
{
    // Try to create the message recursively.
    Message* message = CreateMessageFromService(messageCode, this);
    if(message == NULL)
    {
        message = GetMessageFromTemplate(messageCode);
    }
    return message;
}


bool Transport::Send(const Message* message,
                     const int broadcastFlags) const
{
    Packet::List stream;
    Header::List streamHeaders;
    Packet::List::iterator packet;
    Header::List::iterator header;

    if(message->GetDestinationID() == mComponentID)
    {
        return false;
    }
    // Automatically set the source ID.
    ( (Message * ) message )->SetSourceID(mComponentID);
    mSequenceNumberMutex.Lock();
    if(SerializeMessage(message, stream, streamHeaders, mSequenceNumber, (Byte)broadcastFlags))
    {
        *((UShort *)(&mSequenceNumber)) += (UShort)stream.size();
        mSequenceNumberMutex.Unlock();
        for(packet = stream.begin(), header = streamHeaders.begin();
            packet != stream.end() && header != streamHeaders.end();
            packet++, header++)
        {
            if(SendPacket(*packet, *header, broadcastFlags) == false)
            {
                return false;
            }
        }
        return true;
    }
    mSequenceNumberMutex.Unlock();
    return false;
}


bool Transport::Send(const Message* message, 
                     Message* response, 
                     const unsigned int waitTimeMs) const
{
    Message::List possibleResponses;
    possibleResponses.push_back(response);
    return Send(message, possibleResponses, waitTimeMs);
}


bool Transport::Send(const Message* message, 
                     Message::List& possibleResponses, 
                     const unsigned int waitTimeMs) const
{
    Receipt receipt;
    Receipt::Set::iterator ritr;

    if(message == NULL || possibleResponses.size() == 0 || waitTimeMs == 0 || message->GetDestinationID().IsBroadcast())
    {
        return false;
    }

    mPendingReceiptsMutex.Lock();
    receipt.mPendingFlag = true;
    receipt.mpMessage = message;
    receipt.mpResponses = &possibleResponses;
    receipt.mSendTimeMs = receipt.mUpdateTimeMs = Time::GetUtcTimeMs();
    Receipt::Set* pending = ((Receipt::Set *)&mPendingReceipts);
    pending->insert(&receipt);
    mPendingReceiptsMutex.Unlock();

    if(Send(message))
    {
        Time::Stamp timeMs = Time::GetUtcTimeMs();
        while(timeMs - receipt.mSendTimeMs < waitTimeMs && receipt.mPendingFlag == true)
        {
            timeMs = Time::GetUtcTimeMs();
            // Small sleep statement to not kill processor.
#ifdef WIN32
            CxUtils::SleepMs(1);
#else
            usleep(1);
#endif
        }
    }

    // Regardless of result
    // remove the receipt from the pending set.
    mPendingReceiptsMutex.Lock();
    ritr = pending->find(&receipt);
    if(ritr != pending->end())
    {
        pending->erase(ritr);
    }
    mPendingReceiptsMutex.Unlock();
    
    return !receipt.mPendingFlag;
}


void Transport::RegisterCallback(const UShort messageCode, Transport::Callback* callback)
{
    Mutex::ScopedLock lock(&mCallbacksMutex);
    mMessageCallbacks[messageCode].insert(callback);
}


void Transport::EnableLogging(const bool flag)
{
    Mutex::ScopedLock lock(&mLogFileMutex);
    mLoggingFlag = flag;
    if(flag)
    {
        if(mLogFile.is_open() == false)
        {
            // Generate a file name, and create log file.
            CxUtils::FileIO::CreateDir("logs");
            std::string fname = "logs/JAUS++ Transport Receive Log - ";
            fname += JAUS::Time::GetUtcTime().ToString();
            fname += ".csv";
            mLogFile.open(fname.c_str(), std::ios_base::out);
            if(mLogFile.is_open())
            {
                mLogFile << "Time (UTC), Message Code, Message Name, Source ID, Destination ID, Size (Bytes), Sequence Number, Message Body\n";
                mLogFile.flush();
            }
        }
    }
    else
    {
        if(mLogFile.is_open())
        {
            mLogFile.close();
        }
    }
}


void Transport::ProcessPacket(const Packet& packet, const Header& header)
{
    // Ignore message not destined for this component.
    if(Address::DestinationMatch(header.mDestinationID, mComponentID) == false)
    {
        return;
    }

    if(mLoggingFlag)
    {
        WriteToLog(packet, header);
    }

    if(mMessageProcessingThreads.GetNumThreads() < mNumberOfMessageProcessingThreads)
    {
        char buffer[512];
        sprintf(buffer, "%s_MP_THREAD_%d", mComponentID.ToString().c_str(), mMessageProcessingThreads.GetNumThreads() + 1);
        mMessageProcessingThreads.CreateThread(buffer, Transport::MessageProcessingThread, this);
    }

    UShort messageCode = 0;
    // Read the message code.
    packet.SetReadPos(Header::PayloadOffset);
    packet.Read(messageCode);
    packet.SetReadPos(0);   // Reset the read position.
    
    // Send Acknowledge if requested.
    if(header.mAckNackFlag == Header::AckNack::Request)
    {
        Header ackHeader = header;
        ackHeader.mSourceID = mComponentID;
        ackHeader.mDestinationID = header.mSourceID;
        ackHeader.mSize = Header::MinSize;
        ackHeader.mAckNackFlag = Header::AckNack::Ack;
        ackHeader.mSequenceNumber = header.mSequenceNumber;
        Packet ackPacket;
        if(ackHeader.Write(ackPacket))
        {
            SendPacket(ackPacket, ackHeader);
            if(mLoggingFlag)
            {
                WriteToLog(ackPacket, ackHeader);
            }
        }
    }
    // Check for multi-packet stream
    if(header.mControlFlag == Header::DataControl::Single)
    {
        // Check for blocking receive
        if(CheckPendingReceipts(header, messageCode, packet) == false)
        {
            // Add to queue for processing in message thread.
            mPendingReceiptsMutex.Unlock();
            Mutex::ScopedLock lock(&mJausMessageQueueMutex);
            mJausMessageQueue.PushBack(packet);
        }
    }
    else
    {
        // Multi-packet stream handling.
        UInt presenceVector = 0;
        LargeDataSet::Key key(header.mSourceID,
                              messageCode,
                              presenceVector);

        // Try to find a data set to add to.
        Mutex::ScopedLock lockLargeData(&mLargeDataSetsMutex);
        LargeDataSet::Map::iterator ld;
        ld = mLargeDataSets.find(key);
        bool added = false;

        for(ld = mLargeDataSets.begin();
            ld != mLargeDataSets.end() && added == false;
            ld++)
        {
            if(ld->second->mCompleteFlag == false &&
               ld->first.mMessageCode == key.mMessageCode &&
               ld->first.mSourceID == key.mSourceID)
            {
                // If the first in the sequendce, flush out the old data
                // and start over.
                if(header.mControlFlag == Header::DataControl::First)
                {
                    delete ld->second;
                    mLargeDataSets.erase(ld);
                }                
                else if(ld->second->AddPacket(header, messageCode, packet))
                {
                    added = true;
                }
                break;
            }                   
        }

        if(added == false)
        {
            LargeDataSet* newStream = new LargeDataSet();
            if(newStream->AddPacket(header, messageCode, packet))
            {
                mLargeDataSets[key] = newStream;
            }
            else
            {
                delete newStream;
            }
            newStream = NULL;
        }

        // Check for pending receipts.
        ld = mLargeDataSets.begin();
        while(ld != mLargeDataSets.end())
        {
            if(ld->second->mCompleteFlag == true)
            {
                if(CheckPendingReceipts(ld->second))
                {
                    delete ld->second;
                    ld = mLargeDataSets.begin();
                    continue;
                }
            }
            ld++;
        }
    }
}


Message* Transport::CreateMessageFromService(const UShort messageCode, const Service *service) const
{
    Message* message = NULL;

    if(mStopMessageProcessingFlag)
    {
        return message;
    }

    const Transport* transport = dynamic_cast<const Transport*>(service);
    if(transport)
    {
        message =  GetMessageFromTemplate(messageCode);
    }

    if(message == NULL && this != service)
    {
        message = service->CreateMessage(messageCode);
    }
    if(message == NULL)
    {
        Service::Map children;
        // Copy the child services.
        {
            Mutex::ScopedLock lock(&service->mJausServiceMutex);
            children = service->mJausChildServices;
        }
        Service::Map::const_iterator child;
        for(child = children.begin();
            child != children.end();
            child++)
        {
            if( (message = CreateMessageFromService(messageCode, child->second)) != NULL )
            {
                break;
            }
        }
    }

    return message;
}


Message* Transport::CreateMessageFromPacket(const Packet& packet) const
{
    // Read the header data from the packet.
    packet.SetReadPos(0);
    UShort messageCode = 0;
    packet.Read(messageCode);
    Message* message = CreateMessageFromService(messageCode, this);
    if(message)
    {
        Packet::Wrapper wrapper((unsigned char *)(packet.Ptr() + USHORT_SIZE), packet.Length() - USHORT_SIZE);
        if(message->ReadMessageBody(*wrapper.GetData()) < 0)
        {
            delete message;
            message = NULL;
            if(mDebugMessagesFlag)
            {
                Mutex::ScopedLock lock(&mDebugMessagesMutex);
                std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Received Malformed Packet [0x" << std::setbase(16) << messageCode << std::setbase(10) << "]\n";
            }
        }
    }
    // Reset packet data.
    packet.SetReadPos(0);
    return message;
}


void Transport::TriggerMessageCallbacks(const Message* message)
{
    Mutex::ScopedLock lock(&mCallbacksMutex);
    Callback::Map::iterator cb;
    cb = mMessageCallbacks.find(message->GetMessageCode());
    if(cb != mMessageCallbacks.end())
    {
        Callback::Set::iterator cb2;
        for(cb2 = cb->second.begin();
            cb2 != cb->second.end();
            cb2++)
        {
            (*cb2)->ProcessMessage(message);
        }
    }
}


void Transport::AddMessageTemplate(Message* message)
{
    Mutex::ScopedLock lock(&mMessageTemplatesMutex);
    if(mMessageTemplates.find(message->GetMessageCode()) != mMessageTemplates.end())
    {
        delete mMessageTemplates[message->GetMessageCode()];
    }
    mMessageTemplates[message->GetMessageCode()] = message;
}


Message*  Transport::GetMessageFromTemplate(const UShort messageCode) const
{
    Message* message = NULL;
    {
        Mutex::ScopedLock lock(&mMessageTemplatesMutex);
        std::map<UShort, Message*>::const_iterator m;
        m = mMessageTemplates.find(messageCode);
        if(m != mMessageTemplates.end())
        {
            message = m->second->Clone();
        }
    }
    return message;
}


bool Transport::CheckPendingReceipts(const Header& header, 
                                     const UShort messageCode, 
                                     const Packet& packet)
{
    bool foundReceipt = false;
    Receipt::Set::iterator receipt;
    mPendingReceiptsMutex.Lock();
    for(receipt = mPendingReceipts.begin();
        receipt != mPendingReceipts.end() && false == foundReceipt;
        receipt++)
    {
        Message::List::iterator responses;
        for(responses = (*receipt)->mpResponses->begin();
            responses != (*receipt)->mpResponses->end();
            responses++)
        {              
            if((*responses)->GetMessageCode() == messageCode &&
               (*receipt)->mpMessage->GetDestinationID() == header.mSourceID)
            {
                if( (*responses)->Read(packet) && (*responses)->IsResponseToMessage((*receipt)->mpMessage))
                {
                    foundReceipt = true;
                    (*receipt)->mPendingFlag = false;
                    (*receipt)->mUpdateTimeMs = Time::GetUtcTimeMs();                        
                    Message* final = (*responses);
                    // Remove possible choices so we know which one
                    // has valid data.
                    responses = (*receipt)->mpResponses->begin();
                    while(responses != (*receipt)->mpResponses->end())
                    {
                        if( (*responses) != final )
                        {
                            responses = (*receipt)->mpResponses->erase(responses); continue;
                        }
                        responses++;
                    }
                    mPendingReceipts.erase(receipt);
                    break;
                }
            }
        }
        if(foundReceipt)
            break;
    }
    mPendingReceiptsMutex.Unlock();

    return foundReceipt;
}


bool Transport::CheckPendingReceipts(const LargeDataSet* stream)
{
    bool foundReceipt = false;
    Receipt::Set::iterator receipt;
    mPendingReceiptsMutex.Lock();
    for(receipt = mPendingReceipts.begin();
        receipt != mPendingReceipts.end() && false == foundReceipt;
        receipt++)
    {
        Message::List::iterator responses;
        for(responses = (*receipt)->mpResponses->begin();
            responses != (*receipt)->mpResponses->end();
            responses++)
        {              
            if((*responses)->GetMessageCode() == stream->mMessageCode &&
               (*receipt)->mpMessage->GetDestinationID() == stream->mHeader.mSourceID)
            {
                if( (*responses)->ReadLargeDataSet(stream->mStream) && (*responses)->IsResponseToMessage((*receipt)->mpMessage))
                {
                    foundReceipt = true;
                    (*receipt)->mPendingFlag = false;
                    (*receipt)->mUpdateTimeMs = Time::GetUtcTimeMs();                        
                    Message* final = (*responses);
                    // Remove possible choices so we know which one
                    // has valid data.
                    responses = (*receipt)->mpResponses->begin();
                    while(responses != (*receipt)->mpResponses->end())
                    {
                        if( (*responses) != final )
                        {
                            responses = (*receipt)->mpResponses->erase(responses); continue;
                        }
                        responses++;
                    }
                    mPendingReceipts.erase(receipt);
                    break;
                }
            }
        }
        if(foundReceipt)
            break;
    }
    mPendingReceiptsMutex.Unlock();

    return foundReceipt;
}


void Transport::MessageProcessingThread(void* args)
{
    Transport* transport = (Transport *)args;
    Packet packetToProcess;
    bool dataToProcess = false;
    unsigned int loopCounter = 0;
    
    while(transport && transport->mStopMessageProcessingFlag == false)
    {
        dataToProcess = false;
        packetToProcess.Clear();

        // Grab the next available message.
        // This will block until a message is received.
        transport->mJausMessageQueueMutex.Lock();
        dataToProcess = transport->mJausMessageQueue.PopFront(&packetToProcess);
        transport->mJausMessageQueueMutex.Unlock();
        
        if(dataToProcess && transport->mStopMessageProcessingFlag == false)
        {
            Header header;
            UShort messageCode = 0;
            Message* message = NULL;
            // Read the message type.
            packetToProcess.SetReadPos(0);            
            if(header.Read(packetToProcess) > 0 && 
               packetToProcess.Read(messageCode) > 0)
            {  
                bool foundReceipt = false;
                packetToProcess.SetReadPos(0);                
                // Create message.
                message = transport->CreateMessageFromService(messageCode, transport);
                // If supported, de-serialize data and receive.
                if(message && message->Read(packetToProcess) > 0)
                {
                    if(transport->mDebugMessagesFlag)
                    {
                        Mutex::ScopedLock lock(&transport->mDebugMessagesMutex);
                        std::cout << "[" << transport->GetServiceID().ToString() << "-" << transport->mComponentID.ToString() << "] - Processing " << message->GetMessageName() << " Message\n";
                    }
                    if(transport->mStopMessageProcessingFlag == false)
                    {
                        if(transport->IsPriority(&message) == false)
                        {
                            transport->PushMessageToChildren(message);
                        }
                    }
                }
                else if(message == NULL && transport->mDebugMessagesFlag)
                {
                    Mutex::ScopedLock lock(&transport->mDebugMessagesMutex);
                    std::cout << "[" << transport->GetServiceID().ToString() << "-" << transport->mComponentID.ToString() << "] - Received Unsupported Message Type [0x" << std::setbase(16) << messageCode << std::setbase(10) << "]\n";
                }
            }
            
            // Clean up memory.
            if(message)
            {
                delete message;
            }
        }
        // Check for exit.
        if(transport->mStopMessageProcessingFlag == true)
        {
            break;
        }

        // Check for completed large data sets for processing.
        LargeDataSet* stream = NULL;
        LargeDataSet::Map::iterator lds;        
        transport->mLargeDataSetsMutex.Lock();
        lds = transport->mLargeDataSets.begin();
        while(lds != transport->mLargeDataSets.end())
        {
            // Get completed streams for processing.
            if(lds->second->mCompleteFlag)
            {
                stream = lds->second;
                transport->mLargeDataSets.erase(lds);
                break;
            }
            // Delete streams that have not updated recently.
            else if(Time::GetUtcTimeMs() - lds->second->mUpdateTimeMs > 500)
            {
                delete lds->second;
                transport->mLargeDataSets.erase(lds);
                lds = transport->mLargeDataSets.begin();
                continue;
            }
            lds++;
        }
        transport->mLargeDataSetsMutex.Unlock();

        // Process Multi-Packet Stream.
        if(stream)
        {
            Message* message = NULL;
            message = transport->CreateMessageFromService(stream->mMessageCode, transport);
            if(message && message->ReadLargeDataSet(stream->mStream) && transport->mStopMessageProcessingFlag == false)
            {
                if(transport->IsPriority(&message) == false)
                {
                    transport->PushMessageToChildren(message);
                }
            }
            // Clean up memory.
            if(message)
            {
                delete message;
            }
            delete stream;
            stream = NULL;
        }
        
        if(transport->mMessageProcessingDelayMs == 0)
        {
            // Use a small sleep statement so that we don't
            // max out the CPU, but don't do it all the time.
            #ifdef WIN32
            if( ++loopCounter == 10 )
            {
                loopCounter = 0;
                CxUtils::SleepMs(1);
            }
            #else
            loopCounter = 0;
            usleep(500);
            #endif
        }
        else
        {
            CxUtils::SleepMs(transport->mMessageProcessingDelayMs);
        }
    }

    packetToProcess.Destroy();
}



void Transport::PriorityMessageProcessingThread(void* args)
{
    Transport* transport = (Transport *)args;
    Message::List toProcess;
    Message::List::iterator message;

    unsigned int loopCounter = 0;
    
    while(transport && transport->mStopMessageProcessingFlag == false)
    {
        std::map<Address, Message::Map>::iterator src;
        Message::Map::iterator m;
        transport->mPriorityJausMessageMutex.Lock();
        for(src = transport->mPriorityJausMessages.begin();
            src != transport->mPriorityJausMessages.end();
            src++)
        {
            for(m = src->second.begin();
                m != src->second.end();
                m++)
            {
                toProcess.push_back(m->second);
            }
            src->second.clear();
        }
        transport->mPriorityJausMessageMutex.Unlock();

        for(message = toProcess.begin();
            message != toProcess.end();
            message++)
        {
            transport->PushMessageToChildren(*message);
            delete (*message);
        }
        toProcess.clear();

        if(transport->mMessageProcessingDelayMs == 0)
        {
            // Use a small sleep statement so that we don't
            // max out the CPU, but don't do it all the time.
            #ifdef WIN32
            if( ++loopCounter == 10 )
            {
                loopCounter = 0;
                CxUtils::SleepMs(1);
            }
            #else
            loopCounter = 0;
            usleep(500);
            #endif
        }
        else
        {
            CxUtils::SleepMs(transport->mMessageProcessingDelayMs);
        }
    }

    std::map<Address, Message::Map>::iterator src;
    Message::Map::iterator m;
    transport->mPriorityJausMessageMutex.Lock();
    for(src = transport->mPriorityJausMessages.begin();
        src != transport->mPriorityJausMessages.end();
        src++)
    {
        for(m = src->second.begin();
            m != src->second.end();
            m++)
        {
           delete m->second;
        }
    }
    transport->mPriorityJausMessages.clear();
    transport->mPriorityJausMessageMutex.Unlock();

    for(message = toProcess.begin();
        message != toProcess.end();
        message++)
    {
        delete (*message);
    }
}


void Transport::WriteToLog(const Packet& packet,
                           const Header& header)
{
    if(mLoggingFlag)
    {
        Mutex::ScopedLock lock(&mLogFileMutex);
        if(mLogFile.is_open() == false) { return; }

        std::string messageName = "Unknown";
        mLogFile << Time::GetUtcTime().ToString() << ", ";
        UShort messageCode = 0;
        packet.Read(messageCode, Header::PayloadOffset);
        mLogFile << "0x" << std::hex << std::uppercase << std::setfill('0') << std::setw(4) << messageCode << ", ";
        
        // Lookup message name.
        std::map<UShort, std::string>::iterator n;
        n = mMessageNames.find(messageCode);
        if(n == mMessageNames.end())
        {
            Message* message = CreateMessageFromService(messageCode, this);
            if(message)
            {
                messageName = mMessageNames[messageCode] = message->GetMessageName();
                delete message;
            }
            else
            {
                messageName = "Unknown";
            }
        }
        else
        {
            messageName = n->second;
        }
        mLogFile << messageName << ", ";

        mLogFile << header.mSourceID.ToString() << ", " << header.mDestinationID.ToString() << ", ";
        mLogFile << std::dec << header.mSize << ", " << header.mSequenceNumber << ", ";
        unsigned char* ptr = (unsigned char *)(packet.Ptr() + Header::PayloadOffset + USHORT_SIZE);
        for(unsigned int i = 0; i < packet.Size() - (Header::PayloadOffset + USHORT_SIZE); i++)
        {
            mLogFile << "0x" << std::hex << std::uppercase << std::setfill('0') << std::setw(2) << (unsigned short)*ptr << ", ";
            ptr++;
        }
        mLogFile << std::endl;
        mLogFile.flush();
    }
}


bool Transport::IsPriority(Message** message)
{
    bool result = false;

    JAUS::Event* e = dynamic_cast<JAUS::Event*>(*message);
    UShort messageCode = (*message)->GetMessageCode();
    if(e)
    {
        messageCode = e->GetReportMessageCode();
    }

    mPriorityJausMessageMutex.Lock();
    
    if(mPriorityJausMessageTypes.find(messageCode) != mPriorityJausMessageTypes.end())
    {
        result = true;
        std::map<Address, Message::Map >::iterator src;
        src = mPriorityJausMessages.find((*message)->GetSourceID());
        if(src == mPriorityJausMessages.end())
        {
            mPriorityJausMessages[(*message)->GetSourceID()][(*message)->GetMessageCode()] = (*message);
        }
        else
        {
            Message::Map::iterator msg;
            msg = src->second.find((*message)->GetMessageCode());
            if(msg == src->second.end())
            {
                src->second[(*message)->GetMessageCode()] = (*message);
            }
            else
            {
                delete msg->second;
                msg->second = (*message);
            }
        }
        // De-reference.
        (*message) = NULL;
    }
    mPriorityJausMessageMutex.Unlock();
    return result;
}


/*  End of File */