AddressServer.cpp

Go to the documentation of this file.

////////////////////////////////////////////////////////////////////////////////
/*! \file AddressServer.cpp
*  \brief Defines the class that keeps track of all the networked processes.
*  \author $Author: rsharo $
*  \version $Revision: 1.1 $
*  \date    $Date: 2004/04/07 13:45:04 $
*///////////////////////////////////////////////////////////////////////////////
#include "Comm/AddressServer.h"


AddressServer::AddressServer (u_short port, const char *mcast_addr,
                                                          const char * mcast_iface)
        : m_mcastAddr (port, mcast_addr),
          m_nextObserverID(0)
{
        ACE_INET_Addr in_addr_any;
        m_replySocket.open(in_addr_any);

        if (this->m_mcastDgram.subscribe (this->m_mcastAddr, 1, mcast_iface) == -1)
                ACE_ERROR ((LM_ERROR, "(%N:%l) %p\n", "subscribe"));
}

AddressServer::AddressServer (const ACE_INET_Addr &mcast_addr,
                const char *mcast_iface)
        : m_mcastAddr (mcast_addr),
          m_nextObserverID(0)
{
        ACE_INET_Addr in_addr_any;
        m_replySocket.open(in_addr_any);

        if (this->m_mcastDgram.subscribe (this->m_mcastAddr, 1, mcast_iface) == -1)
                ACE_ERROR ((LM_ERROR, "(%N:%l) %p\n", "subscribe"));
}

AddressServer::~AddressServer (void)
{
        this->m_mcastDgram.close ();
}



int AddressServer::SetLoopbackEnable(bool enabled)
{
        u_char arg = enabled;

        int retVal = m_mcastDgram.set_option(IP_MULTICAST_LOOP, arg);

        if (retVal < 0)
                ACE_ERROR((LM_NOTICE, "(%N:%l): set_option: %p"));

        return retVal;
        
}



int AddressServer::SetTimeToLive(u_char ttl)
{
        int retVal = m_mcastDgram.set_option(IP_MULTICAST_TTL, ttl);

        if (retVal < 0)
                ACE_ERROR((LM_INFO, "(%N:%l): set_option: %p"));

        return retVal;
}



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

        return retVal;
}



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

        return retVal;
}



ACE_HANDLE AddressServer::get_handle (void) const
{
        return this->m_mcastDgram.get_handle ();
}





int AddressServer::handle_input (ACE_HANDLE)
{
        // Receive message from multicast group.
        ssize_t retVal =
                this->m_mcastDgram.recv(m_buf, sizeof(m_buf)-1, m_remoteAddr);

        if (retVal == -1)
        {
                return retVal;
        }
        
        //
        // Null-terminate the string.
        m_buf[retVal] = '\0';
        
        
        //
        // All commands except CLEAR require an object address. This will be it.
        u_int objAddr;

        //
        // The ADD and REMOVE commands require an address argument.
        //
        // Format is <host>:<port>
        //    host is up to MAXHOSTNAMELEN characters
        //    <port> is up to 6 digits ( e.g. 0xABCD )
        //
        // Add colon and null-terminator, max total size is MAXHOSTNAMELEN+8
        char binding[MAXHOSTNAMELEN+8];
        

        //
        // Handle a request for bindings.
        if (sscanf(m_buf, "GET %u", &objAddr) == 1)
        {
                //
                // Prepare to loop through the bindings and write them to 
                // the buffer.

                char *idx = m_buf;
                u_int remaining = sizeof(m_buf);
                
                // Pre-clear the buffer.
                idx[0] = '\0';

                bool firstTime = true;
                BindMap_t::iterator i = m_bindings.find(objAddr);
                
                for(;i != m_bindings.end() && objAddr == i->first; ++i)
                {
                        //
                        // If this is the first time, just fill in the first address.
                        if (firstTime)
                        {
                                firstTime = false;
                        }
                        //
                        // If not the first entry and if there's enough space in the
                        // buffer, insert a comma-separator.
                        else if (remaining > 3)
                        {
                                *(idx++) = ',';
                                --remaining;
                        }
                        //
                        // Else, log a warning and stop adding addresses to the buffer.
                        else
                        {
                                ACE_ERROR((LM_WARNING,
                                        "(%N:%l) Binding list for address %d is too long to send!\n",
                                        objAddr));
                                break;
                        }
                        
                        //
                        // Try to fill in the next binding. If space runs out, then
                        // log a warning and exit the loop.
                        std::string &addr = i->second;
                        if (addr.length() >= remaining)
                        {
                                ACE_ERROR((LM_WARNING,
                                        "(%N:%l) Binding list for address %d is too long to send!\n",
                                        objAddr));
                                break;
                        }
                        //
                        // copy, advance the pointer and keep track of remaining space.
                        else
                        {
                                strcpy(idx, addr.c_str());
                                size_t chunk = strlen(idx);
                                remaining -= chunk;
                                idx += chunk;
                        }
                }

                //
                // Send the addresses. Note that a zero-length message is possible
                // and acceptable for undefined object addresses.
                SendString(m_buf, m_remoteAddr);
        }
        //
        // Handle a request to add a binding.
        else if (sscanf(m_buf, "ADD %u %s", &objAddr, binding) == 2)
        {
                m_bindings.insert(BindValue_t(objAddr, binding));

                SendString("OK", m_remoteAddr);

                NotifyChange(objAddr);
        }
        //
        // Handle a request to remove a binding.
        else if (sscanf(m_buf, "REMOVE %u %s", &objAddr, binding) == 2)
        {
                BindMap_t::iterator i = m_bindings.find(objAddr);
                
                for(;; ++i)
                {
                        if (i == m_bindings.end() || objAddr != i->first)
                        {
                                SendString("NOT_FOUND", m_remoteAddr);
                                break;
                        }

                        std::string &addr = i->second;
                        
                        if (!strcmp(addr.c_str(), binding))
                        {
                                m_bindings.erase(i);

                                SendString("OK", m_remoteAddr);

                                NotifyChange(objAddr);

                                break;
                        }
                }

        }
        //
        // Handle a request to clear the binding list.
        else if (strncmp(m_buf, "CLEAR_BINDINGS", 14) == 0)
        {
                m_bindings.clear();

                SendString("OK", m_remoteAddr);

                NotifyChange();
        }
        //
        // Handle a request for notification on binding change
        else if (strncmp(m_buf, "ADD_NOTIFY", 10) == 0)
        {
                //
                // Handle the (very unlikely) case where every available
                // ID has been assigned.
                if (m_observers.size() == UINT_MAX)
                {
                        SendString("TOO_MANY", m_remoteAddr);
                }
                else
                {
                        //
                        // Handle the (somewhat unlikely) case where IDs have wrapped back
                        // to zero and we have to jump over previously-assigned IDs.
                        NotificationMap_t::iterator i = m_observers.find(m_nextObserverID);
                        while (i != m_observers.end())
                        {
                                m_observers.find(++m_nextObserverID);
                        }
                        
                        //
                        // Add the new observer to the map and tell them what their ID is.
                        m_observers.insert(NotificationValue_t(m_nextObserverID, m_remoteAddr));
                        sprintf(m_buf, "YOU_ARE %u", m_nextObserverID++);
                        SendString(m_buf, m_remoteAddr);
                }
        }
        //
        // Handle a request to end notification on binding change
        else if (sscanf(m_buf, "REMOVE_NOTIFY %u", &objAddr) == 1)
        {
                NotificationMap_t::iterator i = m_observers.find(objAddr);

                if (i == m_observers.end())
                {
                        SendString("BAD_ID", m_remoteAddr);
                }
                else
                {
                        m_observers.erase(i);
                        SendString("OK", m_remoteAddr);
                }
        }
        //
        // Handle a request to clear the notification list.
        else if (strncmp(m_buf, "CLEAR_NOTIFY", 12) == 0)
        {
                m_observers.clear();

                SendString("OK", m_remoteAddr);
        }
        //
        // Respond to a ping.
        else if (strncmp(m_buf, "PING", 4) == 0)
        {
                SendString("ACK", m_remoteAddr);
        }
        else
        {
                ACE_ERROR((LM_WARNING,
                        "(%N:%l) Received unrecognized AddressServer message: %s\n",
                        m_buf));

                SendString("NAK", m_remoteAddr);
                                                                
        }
        
        return 0;
}



int AddressServer::SendString(const char *str, const ACE_INET_Addr &dest)
{
        int retVal = this->m_replySocket.send(str, strlen(str), dest);
        
        if (retVal == -1)
        {
                ACE_ERROR((LM_ERROR, "(%N:%l) %p\n","send"));
        }

        return retVal;
}



int AddressServer::NotifyChange()
{
        int retVal = 0;

        NotificationMap_t::iterator i = m_observers.begin();

        for (; i != m_observers.end(); ++i)
        {
                if (SendString("CHANGED_ALL", i->second) == -1)
                {
                        ACE_ERROR((LM_WARNING,
                                "(%N:%l) Change notification failed for a client. Dropping the client.\n"));
                        m_observers.erase(i);
                        retVal = -1;
                }
        }
        
        return retVal;
}


int AddressServer::NotifyChange(u_int id)
{
        int retVal = 0;

        char buf[25];

        sprintf(buf, "CHANGED %u", id);

        NotificationMap_t::iterator i = m_observers.begin();

        for (; i != m_observers.end(); ++i)
        {
                if (SendString(buf, i->second) == -1)
                {
                        ACE_ERROR((LM_WARNING,
                                "(%N:%l) Change notification failed for a client. Dropping the client.\n"));
                        m_observers.erase(i);
                        retVal = -1;
                }
        }
        
        return retVal;
}

---