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

Go to the documentation of this file.

#include "cxutils/ipc/messageclient.h"
#include "cxutils/time.h"
#include <iostream>
#include <sstream>

using namespace std;
using namespace CxUtils;

unsigned int gMaxQueueSize = 500;

//#define __CX_DEBUG

MessageClient::Callback::Function::Function()
{
    mpCallback = NULL;
    mpCallbackArgs = NULL;
}


MessageClient::Callback::Function::Function(void (*callback)(Packet& packet, void* args),
                                            void* cbArgs)
{
    mpCallback = callback;
    mpCallbackArgs = cbArgs;
}


MessageClient::Callback::Function::Function(const MessageClient::Callback::Function& data)
{
    *this = data;
}


MessageClient::Callback::Function::~Function()
{
}


void MessageClient::Callback::Function::Run(Packet& packet)
{
    if(mpCallback)
    {
        mpCallback(packet, mpCallbackArgs);
    }
}


MessageClient::Callback::Function& MessageClient::Callback::Function::operator=(const MessageClient::Callback::Function& data)
{
    mpCallback = data.mpCallback;
    mpCallbackArgs = data.mpCallbackArgs;
    return *this;
}


MessageClient::MessageClient()
{
    mConnectToServerFlag = false;
    mDesiredBoxSize = 0;
    mUpdateDelayMs = 0;
}


MessageClient::~MessageClient()
{
    Shutdown();
}


int MessageClient::Initialize(const ID serverID, 
                              const ID clientID,
                              const unsigned int messageBoxSize,
                              const bool mustConnect)
{
    std::ostringstream input;
    input << serverID;
    return Initialize(input.str(), clientID, messageBoxSize, mustConnect);
}


int MessageClient::Initialize(const std::string& serverID, 
                              const ID clientID,
                              const unsigned int messageBoxSize,
                              const bool mustConnect)
{
    int result = 0;
    Shutdown();
    
    // Connect to the server.
    if(mRegistry.OpenRegistry(serverID) || mustConnect == false)
    {
        mServerID = serverID;
        mConnectToServerFlag = true;
        mDesiredBoxSize = messageBoxSize;
        // Try to create a message box.
        ID id = clientID;
        if(id == AnyID && mRegistry.IsOpen())
        {
            // Try any available ID.
            for(id = 1; id < 10000 && result == 0; id++)
            {
                if(mRegistry.IsRegistered(id) == false)
                {
                    if(mMessageBox.CreateMessageBox(id, messageBoxSize) || mustConnect == false)
                    {
                        mID = id;
                        if(mReceiveThread.CreateThread(MessageClient::ReceiveThread, this))
                        {
                            //mReceiveThread.SetThreadPriority(50);
                            mReceiveThread.SetThreadName("MessageClientThread");
                            result = 1;
                            break;
                        }
                        else
                        {
                            #ifdef __CX_DEBUG
                            cout << "MessageClient::Initialize - Unable to create thread.\n";
                            #endif
                        }
                    }
                    else
                    {
                        #ifdef __CX_DEBUG
                        cout << "MessageClient::Initialize - Unable to create a Message Box [ID = " << id << "].\n";
                        #endif
                    }
                }
                SleepMs(1);
            }
        }        
        else if(mMessageBox.CreateMessageBox(id, messageBoxSize) || mustConnect == false)
        {
            mID = id;
            if(mReceiveThread.CreateThread(MessageClient::ReceiveThread, this))
            {
                //mReceiveThread.SetThreadPriority(50);
                mReceiveThread.SetThreadName("MessageClientThread");
                result = 1;
            }
            else
            {
                #ifdef __CX_DEBUG
                cout << "MessageClient::Initialize - Unable to create thread.\n";
                #endif
            }
        }
        else
        {
            #ifdef __CX_DEBUG
            cout << "MessageClient::Initialize - Unable to create a Message Box [ID = " << id << "].\n";
            #endif
        }        
    }
    else
    {
        //cout << "MessageClient::Initialize - Unable to open Message Server Registry.\n";
    }
        
    
    
    if(result == 0)
    {
        Shutdown();
    }
    return result;
}


int MessageClient::Initialize(const ID clientID,
                              const unsigned int messageBoxSize)
{
    int result = 0;
    Shutdown();

    if(mMessageBox.CreateMessageBox(clientID, messageBoxSize))
    {
        mID = clientID;
        if(mReceiveThread.CreateThread(MessageClient::ReceiveThread, this))
        {
            //mReceiveThread.SetThreadPriority(50);
            result = 1;
        }
        else
        {
#ifdef __CX_DEBUG
            cout << "MessageClient::Initialize - Unable to create thread.\n";
#endif
        }
    }
    else
    {
#ifdef __CX_DEBUG
        cout << "MessageClient::Initialize - Unable to create a Message Box [ID = " << id << "].\n";
#endif
    }              
    
    if(result == 0)
    {
        Shutdown();
    }
    return result;
}


void MessageClient::Shutdown()
{
    mReceiveThread.StopThread();
    mRegistry.Unregister(mID);
    mRegistry.CloseRegistry();
    mMessageBox.CloseMessageBox();
    mReceivedDataMutex.Lock();
    mReceivedMessages.clear();
    mReceivedDataMutex.Unlock();
    mConnectToServerFlag = false;
    mID = 0;    
    mDesiredBoxSize = 0;
    mServerID.clear();
}



bool MessageClient::IsInitialized() const
{
    return mReceiveThread.IsThreadActive();
}


bool MessageClient::ReadMessage(Packet& message) const
{
    message.Clear();
    Packet::Queue* buffer = ( (Packet::Queue *)(&mReceivedMessages) );
    mReceivedDataMutex.Lock();
    if(buffer->size() > 0)
    {
        message = buffer->front();
        buffer->pop_front();
    }
    mReceivedDataMutex.Unlock();
    return message.Length() > 0 ? true : false;
}


bool MessageClient::ReadMessage(std::string& message) const
{
    message.clear();
    Packet::Queue* buffer = ( (Packet::Queue *)(&mReceivedMessages) );
    mReceivedDataMutex.Lock();
    if(buffer->size() > 0)
    {
        message = (char *)buffer->front().Ptr();
        buffer->pop_front();
    }
    mReceivedDataMutex.Unlock();
    return message.size() > 0 ? true : false;
}


bool MessageClient::RegisterCallback(Callback* callback)
{
    bool result = false;
    std::set<Callback*>::iterator cb;
    mCallbacksMutex.Lock();
    cb = mCallbacks.find(callback);
    if(cb == mCallbacks.end())
    {
        result = true;
        mCallbacks.insert(callback);
    }
    mCallbacksMutex.Unlock();
    return result;
}


bool MessageClient::RegisterCallback(const Callback::Function& callback)
{
    bool result = false;
    std::vector<Callback::Function>::iterator cb;
    mCallbacksMutex.Lock();
    for(cb = mFunctionCallbacks.begin();
        cb != mFunctionCallbacks.end();
        cb++)
    {
        if(cb->mpCallback == callback.mpCallback &&
           cb->mpCallbackArgs == callback.mpCallbackArgs)
           {
               break;
           }
    }
    if(cb == mFunctionCallbacks.end())
    {
        result = true;
        mFunctionCallbacks.push_back(callback);
    }
    mCallbacksMutex.Unlock();
    return result;
}


void MessageClient::RemoveCallback(Callback* callback)
{
    std::set<Callback*>::iterator cb;
    mCallbacksMutex.Lock();
    cb = mCallbacks.find(callback);
    if(cb != mCallbacks.end())
    {
        mCallbacks.erase(callback);
    }
    mCallbacksMutex.Unlock();
}


void MessageClient::RemoveCallback(const Callback::Function& callback)
{
    std::vector<Callback::Function>::iterator cb;
    mCallbacksMutex.Lock();
    for(cb = mFunctionCallbacks.begin();
        cb != mFunctionCallbacks.end();
        cb++)
    {
        if(cb->mpCallback == callback.mpCallback &&
           cb->mpCallbackArgs == callback.mpCallbackArgs)
           {
               mFunctionCallbacks.erase(cb);
               break;
           }
    }
    mCallbacksMutex.Unlock();
}


void MessageClient::SetUpdateDelayMs(const unsigned int delayTimeMs)
{
    Mutex::ScopedLock lock(&mDelayMutex);
    mUpdateDelayMs = delayTimeMs;
}


void MessageClient::ReceiveThread(void* args)
{
    MessageClient* client = (MessageClient *)args;
    #ifdef WIN32
    unsigned int loopCounter = 0;
    #endif
    
    Time::Stamp registryCheckTimeMs = 0;
    
    while(client->mReceiveThread.QuitThreadFlag() == false)
    {
        // Make sure we are registered with the server.
        if(GetTimeMs() - registryCheckTimeMs > 250)
        {
            if(client->mRegistry.IsOpen())
            {
                client->mRegistry.Register(client->mID);
            }
            else if(client->mRegistry.OpenRegistry(client->mServerID))
            {
                ID id = client->mID;
                if(client->mID == AnyID)
                {         
                    id = 1;
                    while(client->mRegistry.IsRegistered(id))
                    {
                        id++;
                        // Don't go forever.
                        if(id - client->mID >= 10000)
                        {
                            break;
                        }
                        //SleepMs(1);
                    }
                }
                if(client->mMessageBox.CreateMessageBox(id, client->mDesiredBoxSize))
                {
                    client->mRegistry.Register(id);
                    client->mID = id;
                }
            }
            registryCheckTimeMs = GetTimeMs();
        }
        
        // Try read a message from the in-box.
        if(client->mMessageBox.ReadMessage(client->mTempPacket))
        {
            // Add to queue, then run callbacks.
            client->mReceivedDataMutex.Lock();
            client->mReceivedMessages.push_back(client->mTempPacket);
            if(client->mReceivedMessages.size() > gMaxQueueSize)
            {
                client->mReceivedMessages.erase(client->mReceivedMessages.begin());
            }
            client->mReceivedDataMutex.Unlock();
            
            // Now run callbacks if we have them.
            client->mCallbacksMutex.Lock();
            
            std::set<Callback*>::iterator cb;
            for(cb = client->mCallbacks.begin();
                cb != client->mCallbacks.end();
                cb++)
            {
                (*cb)->ProcessMessage(client->mTempPacket);
            }
            std::vector<Callback::Function>::iterator cbf;
            for(cbf = client->mFunctionCallbacks.begin();
                cbf != client->mFunctionCallbacks.end();
                cbf++)
            {
                cbf->Run(client->mTempPacket);
            }
            
            client->mCallbacksMutex.Unlock();
        }
        client->mDelayMutex.Lock();
        if(client->mUpdateDelayMs == 0)
        {
            #ifdef WIN32
            if(loopCounter++ == 250)
            {
                loopCounter = 0;
                SleepMs(1);
            }
            #else
            usleep(500);
            #endif
        }
        else
        {
            SleepMs(client->mUpdateDelayMs);
        }
        client->mDelayMutex.Unlock();
    }
    
    // Unregister.
    client->mRegistry.Unregister(client->mID);
}

/*  End of File */