MessageServer.cpp

Go to the documentation of this file.

////////////////////////////////////////////////////////////////////////////////
/*! \file MessageServer.cpp
*  \brief Defines the class that routes messages around the system.
*  \author $Author: rsharo $
*  \version $Revision: 1.1 $
*  \date    $Date: 2004/04/07 13:45:04 $
*///////////////////////////////////////////////////////////////////////////////
#include "Comm/MessageServer.h"
#include "Comm/MessageClient.h"





MessageServer::MessageServer (const ACE_INET_Addr &addrServer, const ACE_INET_Addr &localAddr)
        : m_pAddrClient(new AddressClient(addrServer)),
          m_addrClientOwned(true),
          m_localAddr()
{
        //
        // Open with the specified binding (which could have wildcards).
        m_socket.open(localAddr);

        //
        // Get the complete binding from the socket.
        m_socket.get_local_addr(m_localAddr);
}

MessageServer::MessageServer (AddressClient *addrClient, bool takeOwnership,
                const ACE_INET_Addr &localAddr)
        : m_pAddrClient(addrClient),
          m_addrClientOwned(takeOwnership),
          m_localAddr()
{
        //
        // Open with the specified binding (which could have wildcards).
        m_socket.open(localAddr);

        //
        // Get the complete binding from the socket.
        m_socket.get_local_addr(m_localAddr);
}



MessageServer::~MessageServer (void)
{
        m_socket.close();

        if (m_addrClientOwned)
        {
                delete m_pAddrClient;
        }
}



int MessageServer::RegisterWithReactor(ACE_Reactor &theReactor)
{
        int retVal = theReactor.register_handler (this,
                ACE_Event_Handler::READ_MASK);
                
        if (retVal == -1)
                ACE_ERROR_RETURN((LM_ERROR, "(%N:%l) %p\n", "register_handler"), -1);


        if (m_addrClientOwned)
        {
                retVal = m_pAddrClient->RegisterWithReactor(theReactor);

                if (retVal == -1)
                {
                        theReactor.remove_handler (this, ACE_Event_Handler::READ_MASK);
                }
        }

        return retVal;
}



int MessageServer::RemoveFromReactor(ACE_Reactor &theReactor, bool immediate)
{
        int retVal = theReactor.remove_handler (this,
                ACE_Event_Handler::READ_MASK);
                
        if (retVal == -1)
                ACE_ERROR ((LM_ERROR, "(%N:%l) %p\n", "remove_handler"));

        if (m_addrClientOwned)
        {
                if (m_pAddrClient->RemoveFromReactor(theReactor, immediate) == -1)
                        retVal = -1;
        }

        return retVal;
}



int MessageServer::RegisterClient(u_int id, MessageClient *client)
{
        //
        // NOTE: Insertion of elements does not invalidate iterators,
        // so this method need not be guarded by the mutex.
        //ACE_Guard<ACE_Thread_Mutex> g(m_clientMutex);

        //
        // Check to make sure this client isn't already registered somehow.
        ClientMap_t::iterator it = m_clients.find(id);

        //
        // If this is an entirely new ID, we'll need to bind the address.
        bool addressBindingNeeded = (it == m_clients.end());

        for (; it != m_clients.end() && it->first == id; ++it)
        {
                if (it->second == client)
                {
                        ACE_ERROR_RETURN((LM_ERROR,
                                "(%N:%l) Attempt to register the same client for the same id twice.\n"),
                                -1);
                }
        }

        //
        // Add the client. They will start receiving messages immediately.
        m_clients.insert(ClientValue_t(id, client));
        client->m_pServer = this;

        if (addressBindingNeeded)
        {
                char binding[MAXHOSTNAMELEN+8];
                m_localAddr.addr_to_string(binding, sizeof(binding), 0);
                m_pAddrClient->AddBinding(id, binding);
        }

        return 0;
}


int MessageServer::RemoveClient(u_int id, MessageClient *client)
{
        //
        // Lock out the reactor so we don't corrupt its iterator.
        ACE_Guard<ACE_Thread_Mutex> g(m_clientMutex);

        ClientMap_t::iterator it = m_clients.find(id);
        for (; it != m_clients.end() && it->first == id; ++it)
        {
                if (it->second == client)
                {
                        RemoveClient_i(it);
                        return 0;
                }
        }

        ACE_ERROR_RETURN((LM_ERROR,
                "(%N:%l) Attempt to remove a bad client/id registration.\n"),
                -1);
}



MessageServer::ClientMap_t::iterator
MessageServer::RemoveClient_i(MessageServer::ClientMap_t::iterator it)
{
        //
        // Remember the id of this client.
        u_int id = it->first;

        //
        // Clear the client's server pointer and remove it from the client list.
        it->second->m_pServer = NULL;
        ClientMap_t::iterator next = it;
        ++next;
        m_clients.erase(it);

        //
        // If this was the last client with the given ID, then remove our binding
        // to this ID from the address server.
        ClientMap_t::iterator remain = m_clients.find(id);
        if (remain == m_clients.end())
        {
                char binding[MAXHOSTNAMELEN+8];
                m_localAddr.addr_to_string(binding, sizeof(binding), 0);

                m_pAddrClient->RemoveBinding(id, binding);
        }

        return next;
}


ACE_HANDLE MessageServer::get_handle (void) const
{
        return this->m_socket.get_handle ();
}



int MessageServer::handle_input (ACE_HANDLE)
{
        ACE_INET_Addr remoteAddr;
        MsgHdr_t hdr;

        iovec bufs[2];
        
        bufs[0].iov_len = sizeof(hdr);
        bufs[0].iov_base = (char *)(&hdr);

        bufs[1].iov_len = sizeof(m_buf);
        bufs[1].iov_base = m_buf;


        //
        // Receive message.
        int status =
                this->m_socket.recv(bufs, 2, remoteAddr);

        //
        // If reception failed, log the error and return -1.
        // NOTE: Returning -1 deregisters the server.
        if (status == -1)
        {
                ACE_ERROR_RETURN((LM_ERROR, "(%N:%l) %p\n", "recv"), -1);
        }

        //
        // Check to see if the message was too short...
        // NOTE: This error doesn't deregister the server -- its just a user error.
        if ((u_int)status < sizeof(hdr))
        {
                ACE_ERROR_RETURN((LM_WARNING,
                        "(%N:%l) Received message is shorter than a header. Dropping.\n"),
                        0);
        }
        //
        // Check to see if the header is intact...
        else if (hdr.m_magicNumber == MsgHdr_t::MAGIC_NUMBER)
        {
        }
        //
        // If not intact, see if its just reversed due to an endianness mismatch...
        else if ( hdr.m_magicNumber == ACE_SWAP_LONG(MsgHdr_t::MAGIC_NUMBER) )
        {
                ACE_ERROR((LM_WARNING,
                        "(%N:%l) Sender and receiver don't have the same endianness --"
                        " hope nobody sent any integers...\n"));

                hdr.m_msgLength = ACE_SWAP_LONG(hdr.m_msgLength);
                hdr.m_destination = ACE_SWAP_LONG(hdr.m_destination);
        }
        //
        // Otherwise, its just a bad header.
        // NOTE: This error doesn't deregister the server -- its just a user error.
        else
        {
                ACE_ERROR_RETURN((LM_WARNING,
                        "(%N:%l) Got message with bad/missing header. Dropping.\n"),
                        0);
        }

        //
        // Check to see if the received message length matches what the header
        // tells us it should be...
        // NOTE: This error doesn't deregister the server -- its just a user error.
        size_t actualLen = status - sizeof(hdr);
        if (hdr.m_msgLength != actualLen)
        {
                ACE_ERROR_RETURN((LM_WARNING,
                        "(%N:%l) Message length (%d) is different than header says it"
                        " should be (%d). Dropping.\n",
                        actualLen, hdr.m_msgLength), 0);
        }


        //
        // Prevent removal of clients, since this could corrupt our iterator.
        ACE_Guard<ACE_Thread_Mutex> g(m_clientMutex);

        //
        // Iterate through all the clients and show them this message.
        ClientMap_t::iterator it = m_clients.find(hdr.m_destination);
        while (it != m_clients.end() && it->first == hdr.m_destination)
        {
                //
                // If a client returns -1, then remove them from the server.
                if (it->second->ReceiveMessage(m_buf, actualLen) == -1)
                {
                        it->second->m_pServer = NULL;

                        //
                        // Note that the following line will block until 
                        // the address server's event handler can run --
                        // if the reactor isn't multithreaded this will hang
                        // forever!  Use a thread pool or a WFMO reactor.
                        it = RemoveClient_i(it);
                }
                else
                {
                        ++it;
                }
        }
        
        return 0;
}





int MessageServer::SendMessage(u_int originator, const void *msg, size_t len,
                                                           u_int dest)
{
        int retVal = 0;

        AddrList_t addrList;

        if (m_pAddrClient->GetBindings(dest, addrList) == -1)
        {
                ACE_ERROR_RETURN((LM_ERROR,
                        "(%N:%l) Unable to retrieve bindings from AddressServer.\n"),
                        -1);
        }

        iovec bufs[2];
        
        MsgHdr_t hdr;

        hdr.m_msgLength = len;
        hdr.m_destination = dest;

        bufs[0].iov_len = sizeof(hdr);
        bufs[0].iov_base = (char *)(&hdr);

        bufs[1].iov_len = len;
        bufs[1].iov_base = (char *)(msg);

        AddrList_t::const_iterator i = addrList.begin();

        for (; i != addrList.end(); ++i)
        {
                int status = this->m_socket.send(bufs, 2, *i);
                if (status == -1)
                {
                        ACE_ERROR((LM_ERROR, "(%N:%l) %p\n","send"));
                        retVal = -1;
                }
        }
        
        return retVal;
}

---