C:/Active/libraries/cxutils/2.0/src/cxutils/networking/udpsharedserver.cpp

Go to the documentation of this file.

#include "cxutils/networking/udpsharedserver.h"
#include "cxutils/timer.h"
#include "cxutils/time.h"
#include <iostream>

using namespace std;
using namespace CxUtils;


UdpSharedServer::Info::Info() 
{
    mNetworkInterface = -1;
}


UdpSharedServer::Info::Info(const UdpSharedServer::Info& info) 
{
    *this = info;
}


UdpSharedServer::Info::~Info() 
{
}


UdpSharedServer::Info& UdpSharedServer::Info::operator=(const UdpSharedServer::Info& info) 
{
    mNetworkInterface = info.mNetworkInterface;
    mPort = info.mPort;
    mMulticastGroup = info.mMulticastGroup;
    return *this;
}


UdpSharedServer::Callback::Data::Data()
{
    mpCallback = NULL;
    mpCallbackArgs = NULL;
}


UdpSharedServer::Callback::Data::Data(void (*callback)(const Packet& packet,
                                                       const IP4Address& ipAddress,
                                                       void* args),
                                      void* cbArgs)
{
    mpCallback = callback;
    mpCallbackArgs = cbArgs;
}


UdpSharedServer::Callback::Data::Data(const UdpSharedServer::Callback::Data& data)
{
    *this = data;
}


UdpSharedServer::Callback::Data::~Data()
{
}


void UdpSharedServer::Callback::Data::Run(const Packet& packet, const IP4Address& ipAddress)
{
    if(mpCallback)
    {
        mpCallback(packet, ipAddress, mpCallbackArgs);
    }
}


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


UdpSharedServer::UdpSharedServer() : mPrimaryFlag(false),
                                     mConnectedFlag(false),
                                     mUpdateDelayMs(0),
                                     mpSocket(NULL),
                                     mpMessageServer(NULL),
                                     mpMessageClient(NULL),
                                     mQuitServerFlag(false)
{
    mPacketQueue.Reserve(15);
    mMessageAddressQueue.Reserve(15);
}


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


int UdpSharedServer::InitializeSocket(const unsigned short port, 
                                      const IP4Address& multicastGroup,
                                      const int netInterface)
{
    int result = 0;

    Shutdown();
    mConnectedFlag = false;
    mQuitServerFlag = false;

    mSocketInfo.mPort = port;
    mSocketInfo.mNetworkInterface = netInterface;
    mSocketInfo.mMulticastGroup = multicastGroup;

    // Create thread to initialize appropriate connections.
    if(mUdpStatusThread.CreateThread(UdpSharedServer::UdpStatusThread, this))
    {
        mUdpStatusThread.SetThreadName("UdpSharedServer");
        result = 1;
    }

    // Do cleanup on failure.
    if(result == 0)
    {
        Shutdown();
    }

    return result;
}


void UdpSharedServer::Shutdown()
{
    mQuitServerFlag = true;  
    mUdpStatusThread.StopThread();
    mUdpSocketThread.StopThread();
 
    if(mpMessageServer)
    {
        mpMessageServer->Shutdown();
        delete mpMessageServer;
        mpMessageServer = NULL;
    }
    
    if(mpMessageClient)
    {
        mpMessageClient->Shutdown();
        delete mpMessageClient;
        mpMessageClient = NULL;
    }

    mPrimaryFlag = false;
    mSocketInfo.mMulticastGroup.SetAddress("0.0.0.0");
    mSocketInfo.mNetworkInterface = -1;
    mSocketInfo.mPort = 0;
    
    mQuitServerFlag = false;
    
    if(mpSocket)
    {
        mpSocket->Shutdown();
        delete mpSocket;
        mpSocket = NULL;
    }
    mConnectedFlag = false;
    mPacketQueue.Clear();
    mMessageAddressQueue.Clear();
}


int UdpSharedServer::GetNetworkInterface() const
{
    return mSocketInfo.mNetworkInterface;
}


bool UdpSharedServer::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 UdpSharedServer::RegisterCallback(const Callback::Data& callback)
{
    bool result = false;
    std::vector<Callback::Data>::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 UdpSharedServer::RemoveCallback(Callback* callback)
{
    std::set<Callback*>::iterator cb;
    mCallbacksMutex.Lock();
    cb = mCallbacks.find(callback);
    if(cb != mCallbacks.end())
    {
        mCallbacks.erase(callback);
    }
    mCallbacksMutex.Unlock();
}


void UdpSharedServer::RemoveCallback(const Callback::Data& callback)
{
    std::vector<Callback::Data>::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();
}


bool UdpSharedServer::IsValid() const
{
    if(mUdpStatusThread.IsThreadActive())
        return true;
        
    return false;
}


bool UdpSharedServer::IsPrimary() const
{
    return mPrimaryFlag;
}


bool UdpSharedServer::Recv(Packet& packet, IP4Address& ipAddress)
{
    bool result = false;

    mQueueMutex.Lock();
    if(mPacketQueue.Size() > 0)
    {
        packet = *mPacketQueue.begin();
        ipAddress = *mMessageAddressQueue.begin();
        mPacketQueue.PopFront(NULL);
        mMessageAddressQueue.PopFront(NULL);
        result = true;
    }
    mQueueMutex.Unlock();

    return result;
}


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


void UdpSharedServer::UdpStatusThread(void* udpSharedServer)
{
    UdpSharedServer* server = (UdpSharedServer*)udpSharedServer;
    char messageServerName[512];
    
    sprintf(messageServerName, "%d", server->mSocketInfo.mPort);
            
    while(server && 
          (server->mUdpStatusThread.QuitThreadFlag() == false ||
           server->mQuitServerFlag == false))
    {
        // Check to see if the server is active.
        if(server->mPrimaryFlag == false && server->mpSocket == NULL && MessageServer::IsServerPresent(messageServerName) == false)
        {
            // Try to initialize a UDP Server Socket, if another is present, then this will fail
            server->mpSocket = new UdpServer();
            server->mpSocket->SetNetworkInterface(server->mSocketInfo.mNetworkInterface);
            // Check if this is a multicast socket or not.
            if(server->mSocketInfo.mMulticastGroup.mString >= "224.0.0.0" && 
               server->mSocketInfo.mMulticastGroup.mString <= "239.255.255.255")
            {
                server->mpSocket->InitializeMulticastSocket(server->mSocketInfo.mPort,
                                                            server->mSocketInfo.mMulticastGroup);
            }
            else
            {
                server->mpSocket->InitializeSocket(server->mSocketInfo.mPort);
            }
            
            if(server->mpSocket->IsValid())
            {
                // See if we can take ownership as the primary
                server->mpMessageServer = new MessageServer();
                // This will only succeed if there is no active other server.
                if(server->mpMessageServer->Initialize(messageServerName, MessageServer::DefaultMaxClients))
                {
                    server->mPrimaryFlag = true; // SUCCESS!
                    // If we were a client, shutdown interface
                    if(server->mpMessageClient)
                    {
                        delete server->mpMessageClient;
                        server->mpMessageClient = NULL;
                        
                    }
                    // Create a thread to receive data from the socket.
                    server->mUdpSocketThread.CreateThread(UdpSharedServer::UdpSocketThread, server);
                    //server->mUdpSocketThread.SetThreadPriority(50);
                    server->mUdpSocketThread.SetThreadName("UdpSocketServer");      
                }
            }

            if(server->mPrimaryFlag == false)
            {
                // Shutdown and delete servers
                if(server->mpSocket)
                {
                    delete server->mpSocket;
                }
                if(server->mpMessageServer)
                {
                    delete server->mpMessageServer;
                }
                server->mpSocket = NULL;
                server->mpMessageServer = NULL;
            }
        }
        // Make sure we have an active shared memory client interface running
        // otherwise we won't receive any messages!
        if(server->mPrimaryFlag == false && server->mpMessageClient == NULL)
        {
            server->mpMessageClient = new MessageClient();
            server->mpMessageClient->SetUpdateDelayMs(server->mUpdateDelayMs);
            if(server->mpMessageClient->Initialize(messageServerName, MessageClient::AnyID) == 0)
            {
                delete server->mpMessageClient;
                server->mpMessageClient = NULL;
            }
            else
            {
                // Register callback to get messages as they arrive
                server->mpMessageClient->RegisterCallback(MessageClient::Callback::Function(UdpSharedServer::ProcessUdpMessage, server));
            }
        }

        SleepMs(250);
    }
}


void UdpSharedServer::UdpSocketThread(void* udpSharedServer)
{
    UdpSharedServer* server = (UdpSharedServer*)udpSharedServer;
    Packet udpMessage;
    IP4Address sourceAddress;
    unsigned short sourcePort = 0;
#ifdef WIN32
    unsigned int loopCounter = 0;
#endif

    while(server && 
          (server->mUdpSocketThread.QuitThreadFlag() == false ||
           server->mQuitServerFlag == false))
    {
        // Try and read a new incomming message.
        udpMessage.Clear(false);
        sourcePort = 0;
        if(server->mpSocket->Recv(udpMessage, 5000, 100, &sourceAddress, &sourcePort) > 0)
        {
            // Add the IP address to the end of the message for shared memory clients.
            udpMessage.Write(sourceAddress.mData, 4, udpMessage.Length());
            udpMessage.Write(sourcePort, udpMessage.Length());
            // Send to non-primary UdpSharedServer instances.
            server->mpMessageServer->SendToAllClients(udpMessage);
            // Now process the data internally.  This will add to the
            // queue of received messages, and generate callbacks.
            UdpSharedServer::ProcessUdpMessage(udpMessage, server);
        }
        server->mDelayMutex.Lock();
        if(server->mUpdateDelayMs == 0)
        {
#ifdef WIN32
            // Only sleep every N loops
            if( loopCounter++ == 250)
            {
                loopCounter = 0;
                SleepMs(1);
            }
#else
            usleep(500);
#endif
        }
        else
        {
            CxUtils::SleepMs(server->mUpdateDelayMs);
        }
        server->mDelayMutex.Unlock();
    }
    server->mConnectedFlag = false;
}


void UdpSharedServer::ProcessUdpMessage(Packet& udpMessage, void* udpSharedServer)
{
    UdpSharedServer* server = (UdpSharedServer*)udpSharedServer;
    IP4Address sourceAddress;
    unsigned short sourcePort = 0;
    // Read out the ipAddress data from the packet.
    const unsigned char * ptr = udpMessage.Ptr() + udpMessage.Length() - 6;
    sourceAddress.SetAddress(*ptr, *(ptr + 1), *(ptr + 2), *(ptr + 3));
    udpMessage.Read(sourcePort, udpMessage.Length() - 2);
    udpMessage.Write((unsigned char)0, (unsigned int)udpMessage.Length() - 6);
    udpMessage.SetLength(udpMessage.Length() - 6);
    
    server->mConnectedFlag = true;

    // Generate callbacks.
    std::set<Callback*>::iterator cb;
    server->mCallbacksMutex.Lock();
    for(cb = server->mCallbacks.begin();
        cb != server->mCallbacks.end();
        cb++)
    {
        (*cb)->ProcessUDP(udpMessage, sourceAddress);
        (*cb)->ProcessUDP(udpMessage, sourceAddress, sourcePort);
    }
    
    std::vector<Callback::Data>::iterator cbf;
    for(cbf = server->mFunctionCallbacks.begin();
        cbf != server->mFunctionCallbacks.end();
        cbf++)
    {
        cbf->Run(udpMessage, sourceAddress);
    }
    
    server->mCallbacksMutex.Unlock();

    // Add to queue
    server->mQueueMutex.Lock();
    server->mPacketQueue.PushBack(udpMessage);
    server->mMessageAddressQueue.PushBack(sourceAddress);
    server->mQueueMutex.Unlock();
    // Done!
}

/*  End of File */