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

Go to the documentation of this file.

#include "cxutils/ipc/messageserver.h"
#include <iostream>
#include <string.h>
#include <sstream>
#include <cstdio>

using namespace std;
using namespace CxUtils;

unsigned int gInactiveThreshMs = 250;

//#define __CX_DEBUG

MessageServer::Registry::Registry()
{
    mServerFlag = false;
}


MessageServer::Registry::~Registry()
{
    CloseRegistry();
}


int MessageServer::Registry::OpenRegistry(const ID registryID)
{
    std::ostringstream input;
    input << registryID;
    return OpenRegistry(input.str());
}


int MessageServer::Registry::OpenRegistry(const std::string& registryID)
{
    int result = 0;
    CloseRegistry();

    if(mRegistry.OpenMappedMemory(registryID))
    {
        mRegistry.Lock();
        // Check for valid header.
        if(memcmp(mRegistry->Ptr(), "REGISTRY", 8) == 0)
        {
            result = 1;
        }
        mRegistry.Unlock();
    }
    if(result == 0)
    {
        mRegistry.CloseMappedMemory();
    }
    return result;
}


int MessageServer::Registry::CloseRegistry()
{
    List registry;
    if(mRegistry.IsOpen())
    {
        std::string name;
        name = mRegistry.GetMappedName();
        mRegistry.Lock();

        // Make sure mapped memory is deleted.
        unsigned int count = 0;
        // Get the size of the registry.
        mRegistry->SetReadPos(CountPos);
        mRegistry->Read(count);
        if(count == 0 && mServerFlag == true)
        {
            // We are the last to close a view of
            // this, so flush contents.
            mRegistry->SetWritePos(0);
            mRegistry->Write( std::string("REGISTRY") );
            mRegistry->Write( ( (unsigned int)0 ) );
            mRegistry->Write( ( (unsigned int)0 ) );
        }

        mRegistry.Unlock();
        mRegistry.CloseMappedMemory();
    }

    return 1;
}


int MessageServer::Registry::Register(const ID id)
{
    int result = 0;
    if(mRegistry.IsOpen())
    {
        unsigned int count = 0;
        ID temp = 0;
        mRegistry.Lock();

        // Get the contents of the registry.
        mRegistry->SetReadPos(CountPos);
        mRegistry->Read(count);
        for(unsigned int i = 0; i < count; i++)
        {
            mRegistry->Read(temp);
            if(temp == id)
            {
                result = 1;
                break;
            }
        }
        if(result == 0)
        {
            mRegistry->SetWritePos(HeaderSize + count*sizeof(ID));
            if(mRegistry->Write(id))
            {
                count++;
                mRegistry->SetWritePos(CountPos);
                mRegistry->Write(count);
                result = 1;
            }
        }
        mRegistry.Unlock();
    }

    return result;
}


int MessageServer::Registry::Unregister(const ID id)
{
    int result = 0;
    if(mRegistry.IsOpen())
    {
        unsigned int count = 0;
        ID temp = 0;
        mRegistry.Lock();

        // Get the contents of the registry.
        mRegistry->SetReadPos(CountPos);
        mRegistry->Read(count);
        for(unsigned int i = 0; i < count; i++)
        {
            mRegistry->Read(temp);
            if(temp == id)
            {
                result = 1;
                count--;
                memmove(mRegistry->Ptr() + HeaderSize + i*sizeof(ID),
                        mRegistry->Ptr() + HeaderSize + (i+1)*sizeof(ID),
                        count*sizeof(ID));
                break;
            }
        }
        if(result == 1)
        {
            mRegistry->SetWritePos(CountPos);
            mRegistry->Write(count);
        }
        mRegistry.Unlock();
    }
    return result;
}


int MessageServer::Registry::GetRegistry(MessageServer::List& registry)
{
    int result = 0;
    registry.clear();

    if(mRegistry.IsOpen())
    {
        result = 1;
        unsigned int count = 0;
        ID temp = 0;
        mRegistry.Lock();

        // Get the contents of the registry.
        mRegistry->SetReadPos(CountPos);
        mRegistry->Read(count);
        for(unsigned int i = 0; i < count; i++)
        {
            mRegistry->Read(temp);
            registry.insert(temp);
        }

        mRegistry.Unlock();
    }
    return result;
}


bool MessageServer::Registry::IsRegistered(const ID id) const
{
    bool result = false;
    if(mRegistry.IsOpen())
    {
        unsigned int count = 0;
        ID temp = 0;
        mRegistry.Lock();

        // Get the contents of the registry.
        mRegistry->SetReadPos(CountPos);
        mRegistry->Read(count);
        for(unsigned int i = 0; i < count; i++)
        {
            mRegistry->Read(temp);
            if(temp == id)
            {
                result = true;
                break;
            }
        }
        mRegistry.Unlock();
    }
    return result;
}


bool MessageServer::Registry::IsActive(const unsigned int thresholdMs) const
{
    bool result = false;

    if(mRegistry.IsOpen())
    {
        mRegistry.Lock();
        // Check to see if the registry is being used
        // by a server.
        mRegistry->SetReadPos(TimePos);
        Time::Stamp updateTimeMs = 0;
        if(mRegistry->Read(updateTimeMs))
        {
            if((unsigned int)(GetTimeMs() - updateTimeMs) <= thresholdMs)
            {
                result = true;
            }
        }
        mRegistry.Unlock();
    }

    return result;
}


int MessageServer::Registry::CreateRegistry(const ID registryID,
                                            const unsigned int maxClients)
{
    std::ostringstream input;
    input << registryID;
    return CreateRegistry(input.str(), maxClients);
}


int MessageServer::Registry::CreateRegistry(const std::string& registryID,
                                            const unsigned int maxClients)
{
    int result = 0;
    unsigned int size = maxClients*sizeof(ID) + HeaderSize;

    CloseRegistry();

    if(mRegistry.OpenMappedMemory(registryID))
    {
        mRegistry.Lock();
        // Check for valid header.
        if(memcmp(mRegistry->Ptr(), "REGISTRY", 8) == 0)
        {
            // Check to see if the registry is already being used
            // by another server.
            mRegistry->SetReadPos(TimePos);
            Time::Stamp updateTimeMs = 0;
            if(mRegistry->Read(updateTimeMs))
            {
                if(GetTimeMs() - updateTimeMs > 500)
                {
                    result = 1;
                    mRegistry->SetWritePos(TimePos);
                    // Set a new write time.
                    mRegistry->Write( GetTimeMs() );
                }
            }
        }
        else
        {
            mRegistry->SetWritePos(0);
            mRegistry->Write( std::string("REGISTRY") );
            mRegistry->Write( GetTimeMs() );
            result = 1;
        }
        mRegistry.Unlock();
    }
    else if(mRegistry.CreateMappedMemory(registryID, size))
    {
        mRegistry.Lock();
        mRegistry->SetWritePos(0);
        mRegistry->Write( std::string("REGISTRY") );
        mRegistry->Write( GetTimeMs() );
        mRegistry->Write( ( (unsigned int)0 ) );
        mRegistry.Unlock();
        result = 1;
    }

    if(result == 0)
    {
        mRegistry.CloseMappedMemory();
    }
    else
    {
        mServerFlag = true;
    }
    return result;
}


bool MessageServer::Registry::Touch(const Time::Stamp timestampMs)
{
    if(mRegistry.IsOpen())
    {
        mRegistry.Lock();
        mRegistry->SetWritePos(TimePos);
        // Set a new write time.
        mRegistry->Write( timestampMs );
        mRegistry.Unlock();
        return true;
    }
    return false;
}


MessageServer::MessageServer()
{
}


MessageServer::~MessageServer()
{
}


int MessageServer::Initialize(const ID serverID,
                              const unsigned int maxClients)
{
    std::ostringstream input;
    input << serverID;
    return Initialize(input.str(), maxClients);
}


int MessageServer::Initialize(const std::string& serverID,
                              const unsigned int maxClients)
{
    int result = 0;

    // Close if already running.
    Shutdown();

    if(mClientRegistry.CreateRegistry(serverID, maxClients))
    {
        if(mClientDiscoveryThread.CreateThread(MessageServer::ClientDiscoveryThread, this))
        {
            result = 1;
        }
        else
        {
            cout << "MessageServer::Initialize - Failed to create thread.\n";
        }
    }
    else
    {
#ifdef __CX_DEBUG
        cout << "MessageServer::Initialize - Failed to create registry for clients.\n";
        cout << "Another server may be running, or a previouis instance of an application\nrunning a server didn't close properly (delete memory or reboot system to force release).\n";
#endif
    }

    if(result == 0)
    {
        Shutdown();
    }

    return result;
}


void MessageServer::Shutdown()
{
    mClientDiscoveryThread.StopThread();

    if(mClientRegistry.IsOpen())
    {
        // Write a 0 value so another server
        // can claim ownership.
        mClientRegistry.Touch(0);
        mClientRegistry.CloseRegistry();
    }
    // Delete allocated memory.
    Clients::iterator client;
    for(client = mClients.begin();
        client != mClients.end();
        client++)
    {
        client->second->CloseMessageBox();
        delete client->second;
    }
    mClients.clear();
}


bool MessageServer::CreateConnection(const ID id)
{
    bool result = false;
    MappedMessageBox* client = new MappedMessageBox();

    if(client->OpenMessageBox(id))
    {
        mClientRegistry.Register(id);
        Clients::iterator c;
        mClientsMutex.Lock();
        c = mClients.find(id);
        if(c != mClients.end())
        {
            c->second->CloseMessageBox();
            delete c->second;
            mClients.erase(c);
        }
        mClients[id] = client;
        client = NULL;
        mClientsMutex.Unlock();
        result = true;
    }

    if(result == false && client)
    {
        delete client;
        client = NULL;
    }

    return result;
}

bool MessageServer::SendToClient(const ID id, const Packet& message)
{
    bool result = false;
    Clients::iterator client;
    mClientsMutex.Lock();
    client = mClients.find(id);
    if(client != mClients.end())
    {
        result = client->second->WriteMessage(message);
    }
    mClientsMutex.Unlock();
    return result;
}


bool MessageServer::SendToClient(const ID id, const std::string& message)
{
    bool result = false;
    Clients::iterator client;
    mClientsMutex.Lock();
    client = mClients.find(id);
    if(client != mClients.end())
    {
        result = client->second->WriteMessage(message);
    }
    mClientsMutex.Unlock();
    return result;
}


bool MessageServer::SendToAllClients(const Packet& message)
{
    bool result = false;
    Clients::iterator client;
    mClientsMutex.Lock();
    for(client = mClients.begin();
        client != mClients.end();
        client++)
    {
        result = client->second->WriteMessage(message);
    }
    mClientsMutex.Unlock();
    return result;
}


bool MessageServer::SendToAllClients(const std::string& message)
{
    bool result = false;
    Clients::iterator client;
    mClientsMutex.Lock();
    for(client = mClients.begin();
        client != mClients.end();
        client++)
    {
        result = client->second->WriteMessage(message);
    }
    mClientsMutex.Unlock();
    return result;
}


bool MessageServer::GetClientList(List& clients) const
{
    bool result = false;
    Clients::const_iterator client;
    clients.clear();
    mClientsMutex.Lock();
    for(client = mClients.begin();
        client != mClients.end();
        client++)
    {
        result = true;
        clients.insert(client->first);
    }
    mClientsMutex.Unlock();
    return result;
}


unsigned int MessageServer::GetNumClients() const
{
    unsigned int result = 0;
    mClientsMutex.Lock();
    result = (unsigned int)mClients.size();
    mClientsMutex.Unlock();
    return result;
}


bool MessageServer::IsInitialized() const
{
    if(mClientRegistry.IsOpen() && mClientDiscoveryThread.IsThreadActive())
    {
        return true;
    }
    return false;
}


bool MessageServer::IsServerPresent(const std::string& serverID)
{
    bool result = false;
    MappedMemory registry;
    if(registry.OpenMappedMemory(serverID))
    {
        registry.Lock();
        // Check for valid header.
        if(memcmp(registry->Ptr(), "REGISTRY", 8) == 0)
        {
            // Check the timestamp to see if it is being written to.
            registry->SetReadPos(Registry::TimePos);
            Time::Stamp updateTimeMs = 0;
            if(registry->Read(updateTimeMs))
            {
                if(GetTimeMs() - updateTimeMs <= 500)
                {
                    result = true;
                }
            }
        }
        registry.Unlock();
    }

    return result;
}

void MessageServer::ClientDiscoveryThread(void* args)
{
    MessageServer* server = (MessageServer *)args;
    MessageServer::List registry;
    MessageServer::List myRegistry;
    MessageServer::List::iterator client;
    Clients::iterator connection;

    while(server->mClientDiscoveryThread.QuitThreadFlag() == false)
    {
        // Get a copy of the current registry.
        if(server->mClientRegistry.IsOpen() && server->mClientRegistry.GetRegistry(registry))
        {
            // Create new client connections.
            for(client = registry.begin();
                client != registry.end();
                client++)
            {
                // Check for new connection.
                if(myRegistry.find( *client ) == myRegistry.end())
                {
                    server->mClientsMutex.Lock();
                    connection = server->mClients.find(*client);
                    if(connection != server->mClients.end())
                    {
                        // Have a connection already.
                        myRegistry.insert(*client);
                    }
                    else
                    {
                        // Try create a new connection.
                        MappedMessageBox* newConnection = new MappedMessageBox();
                        char name[256];
                        sprintf(name, "%llu", *client);
                        if(newConnection->OpenMessageBox(name))
                        {
                            myRegistry.insert(*client);
                            server->mClients[*client] = newConnection;
                        }
                        else
                        {
                            delete newConnection;
                            // Remove from registry since we cannot connect
                            // to them (probably an inactive block of shared memory).
                            server->mClientRegistry.Unregister(*client);
                        }
                    }
                    server->mClientsMutex.Unlock();
                }
            }
            // Check for disconnect event.
            client = myRegistry.begin();
            while(client != myRegistry.end())
            {
                // If not found in current registry, then we
                // need to close our connection.
                if(registry.find(*client) == registry.end())
                {
                    server->mClientsMutex.Lock();
                    connection = server->mClients.find(*client);
                    if(connection != server->mClients.end())
                    {
                        connection->second->CloseMessageBox();
                        delete connection->second;
                        server->mClients.erase(connection);
                    }
                    server->mClientsMutex.Unlock();
                    // Remove from current list.
                    myRegistry.erase(client);
                    client = myRegistry.begin();
                }
                else
                {
                    client++;
                }
            }
        }

        // Check for in-active client connections and delete them.
        server->mClientsMutex.Lock();
        client = myRegistry.begin();
        while(client != myRegistry.end())
        {
            connection = server->mClients.find( *client );
            if(connection == server->mClients.end())
            {
                myRegistry.erase(client);
                client = myRegistry.begin();
            }
            else
            {
                client++;
            }
        }
        connection = server->mClients.begin();
        while(connection != server->mClients.end())
        {
            if(connection->second->IsActive(true, gInactiveThreshMs) == false)
            {
                connection->second->CloseMessageBox();
                delete connection->second;
                server->mClients.erase(connection);
                connection = server->mClients.begin();
            }
            else
            {
                connection++;
            }
        }
        server->mClientsMutex.Unlock();

        // Indicate that we are reading the registry by
        // updating a time stamp in it that only the server
        // can access.
        server->mClientRegistry.Touch(GetTimeMs());

        SleepMs(100);
    }
}

/*  End of File */