AddressClient.cpp

Go to the documentation of this file.

////////////////////////////////////////////////////////////////////////////////
/*! \file AddressClient.cpp
*  \brief Defines the class that communicates with an address server.
*  \author $Author: rsharo $
*  \version $Revision: 1.1 $
*  \date    $Date: 2004/04/07 13:45:03 $
*///////////////////////////////////////////////////////////////////////////////
#include "Comm/AddressClient.h"

#include <string>

AddressClient::AddressClient (const ACE_INET_Addr &addrServer)
        : m_addrServer(addrServer),
          m_bindingMutex(),
          m_bindings(),
          m_pendingReplyMutex(),
          m_pendingReplyQueue(m_pendingReplyMutex),
          m_notificationID(-1),
          m_pingMutex(),
          m_pingCondition(m_pingMutex),
          m_timeout(0,0)
{
        m_socket.open(ACE_INET_Addr());
}



AddressClient::~AddressClient (void)
{
        m_socket.close();
}


int AddressClient::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 AddressClient::RemoveFromReactor(ACE_Reactor &theReactor, bool immediate)
{
        if (!immediate)
        {
                RemoveServerCallback();
        }

        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;
}



void AddressClient::SetTimeout(const ACE_Time_Value &tv)
{
        m_timeout = tv;
}



int AddressClient::PingServer()
{
        //
        // Unlike other commands that use the m_timeout member to determine
        // how long to wait, PingServer() always waits two seconds.  This should
        // be plenty of time and prevents deadlock when m_timeout is zero (wait
        // forever).
        static const ACE_Time_Value PING_TIMEOUT(1,0);

        //
        // Acquire the mutex to lock out other callers.
        ACE_Guard<ACE_Thread_Mutex> g(m_pingMutex);

        //
        // Try to send the ping command to the address server.
        int retVal = this->m_socket.send("PING", 4, m_addrServer);
        if (retVal == -1)
        {
                ACE_ERROR_RETURN((LM_ERROR, "(%N:%l) %p\n","send"), -1);
        }

        //
        // Set the deadline for one second from now.
        ACE_Time_Value deadline = ACE_OS::gettimeofday() + PING_TIMEOUT;

        //
        // Return 0 if handle_input() reports an "ACK" response,
        // -1 if we timeout or experience an error.
        return m_pingCondition.wait(&deadline);
}



int AddressClient::RegisterServerCallback()
{
        //
        // If we are already registered, don't try to register again.
        if (m_notificationID != -1)
        {
                return -1;
        }

        //
        // First, make sure the server is there so we don't deadlock.
        int status = -1;
        for (int i = 0; status == -1 && i < 1; ++i)
        {
                status = PingServer();
        }

        //
        // If we can't get an answer from the server, then give up.
        if (status == -1)
        {
                ACE_ERROR_RETURN((LM_ERROR,
                        "(%N:%l) RegisterServerCallback failed because the address server"
                        " won't respond.\n"), -1);
        }

        //
        // Send the command and wait for reply.
        std::string theReply;
        status = SendCommand("ADD_NOTIFY", theReply);

        //
        // Try to get our id from the reply.
        u_int myID;
        if (status == -1 || sscanf(theReply.c_str(), "YOU_ARE %u", &myID) != 1)
        {
                return -1;
        }

        //
        // Store our id.
        m_notificationID = myID;

        return 0;
}



int AddressClient::RemoveServerCallback()
{
        //
        // If we're aren't registered, then don't try to remove.
        if (m_notificationID == -1)
        {
                return -1;
        }

        //
        // First, make sure the server is there so we don't deadlock.
        int status = -1;
        for (int i = 0; status == -1 && i < 5; ++i)
        {
                status = PingServer();
        }

        //
        // If we can't get an answer from the server, then give up.
        if (status == -1)
        {
                ACE_ERROR_RETURN((LM_WARNING,
                        "(%N:%l) RemoveServerCallback failed because the address server"
                        " won't respond.\n"), -1);
        }

        //
        // We'll be sending the "REMOVE_NOTIFY" command with our ID.
        u_int id = m_notificationID;
        char sendBuf[32];
        sprintf(sendBuf, "REMOVE_NOTIFY %u", id);

        //
        // Send the command to the address server and wait for the reply.
        std::string theReply;
        status = SendCommand(sendBuf, theReply);

        //
        // If the command failed, return an error.
        if (status == -1 || theReply != "OK")
        {
                return -1;
        }

        //
        // Clear our id.
        m_notificationID = -1;

        return 0;
}


int AddressClient::ClearServerCallbacks()
{
        int retVal = -1;

        std::string theReply;
        retVal = SendCommand("CLEAR_NOTIFY", theReply);

        if (theReply == "OK")
        {
                m_notificationID = -1;
                retVal = 0;
        }

        return retVal;
}



int AddressClient::GetBindings(u_int dest, AddrList_t &theList)
{

        //
        // Try to get the binding from the cache.
        {
                ACE_Guard<ACE_Thread_Mutex> g(m_bindingMutex);

                BindMap_t::iterator it = m_bindings.find(dest);

                if (it != m_bindings.end())
                {
                        theList = it->second;
                        return 0;
                }
        }

        
        //
        // We didn't get the binding from the cache. Try to get it
        // from the address server (and write it to the cache at the
        // same time).
        //
        // Send a "GET" command to the address server.
        char sendBuf[25];
        sprintf(sendBuf, "GET %u", dest);
        
        std::string theReply;
        if (SendCommand(sendBuf, theReply) == -1)
        {
                return -1;
        }
        
        
        if (theReply == "NAK")
        {
                return -1;
        }
        
        theList.clear();
        
        //
        // Iterate through the reply and fill the vector with the bindings.
        if (theReply.size() > 0)
        {
                //
                // Scan backward through the comma-separated entries.
                size_t comma = theReply.find_last_of(',');
                
                while (comma != std::string::npos)
                {
                        //
                        // The binding is everything following the comma.
                        const char *binding = theReply.c_str() + (comma + 1);
                        
                        //
                        // Store the binding
                        theList.push_back(ACE_INET_Addr(binding));
                        
                        //
                        // Remove the comma and everything following it.
                        theReply.erase(comma);
                        
                        //
                        // Prepare for next iteration.
                        comma = theReply.find_last_of(',');
                }
                
                // Store the final binding.
                theList.push_back(ACE_INET_Addr(theReply.c_str()));
        }
        
        {
                ACE_Guard<ACE_Thread_Mutex> g(m_bindingMutex);
                
                //
                // Insert an entry into the bindings and assign the list into it.
                // By passing an empty address list and then
                // assigning it, we save one copy construction.
                BindMap_t::iterator it = m_bindings.insert( BindValue_t(dest, AddrList_t()) ).first;
                it->second = theList;
        }

        return 0;
}



int AddressClient::AddBinding(u_int id, const char *binding)
{
        int retVal = -1;

        char sendBuf[25];
        sprintf(sendBuf, "ADD %u %s", id, binding);

        std::string theReply;
        if (SendCommand(sendBuf, theReply) == -1)
        {
                return -1;
        }

        if (theReply == "OK")
        {
                retVal = 0;
        }

        return retVal;
}



int AddressClient::RemoveBinding(u_int id, const char *binding)
{
        int retVal = -1;

        char sendBuf[25];
        sprintf(sendBuf, "REMOVE %u %s", id, binding);

        std::string theReply;
        if (SendCommand(sendBuf, theReply) == -1)
        {
                return -1;
        }

        if (theReply == "OK")
        {
                retVal = 0;
        }

        return retVal;
}



int AddressClient::ClearBindings()
{
        int retVal = -1;

        std::string theReply;
        if (SendCommand("CLEAR_BINDINGS", theReply) == -1)
        {
                return -1;
        }

        if (theReply == "OK")
        {
                retVal = 0;
        }

        return retVal;
}




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



int AddressClient::handle_input (ACE_HANDLE)
{
        ACE_INET_Addr remoteAddr;

        int len = m_socket.recv(m_buf, sizeof(m_buf)-1, remoteAddr);

        //
        // If the receive call failed, report the error and
        // return -1.
        // NOTE: A return value of -1 means the AddressClient is 
        // deregistered with the Reactor.
        if (len == -1)
        {
                ACE_ERROR_RETURN((LM_ERROR, "(%N:%l) %p\n", "recv"), -1);
        }

        //
        // Null-terminate the received string.
        m_buf[len] = '\0';

        //
        // This object address is needed for "CHANGED" commands.
        u_int objAddr;

        //
        // If the received message is "CHANGED_ALL" then all the cached
        // bindings are invalidated.
        if (strncmp("CHANGED_ALL", m_buf, 11) == 0)
        {
                ACE_DEBUG((LM_INFO,
                        "(%N:%l) Address server changed. Clearing cache.\n"));

                ACE_Guard<ACE_Thread_Mutex> g(m_bindingMutex);
                
                //
                // Clear all the bindings.
                m_bindings.clear();
        }
        //
        // If the received message is "CHANGED" then the cached bindings
        // with the given object address are invalidated.
        else if (sscanf(m_buf, "CHANGED %u", &objAddr) == 1)
        {
                ACE_DEBUG((LM_INFO,
                        "(%N:%l) Address server changed. Clearing entries for %u.\n",
                        objAddr));

                ACE_Guard<ACE_Thread_Mutex> g(m_bindingMutex);
                
                //
                // If we have bindings for this address, remove them.
                m_bindings.erase(objAddr);
        }
        //
        // If the message is "ACK", then someone pinged the server and wants to
        // know its alive. Tell them it is.
        else if (strncmp("ACK", m_buf, 3) == 0)
        {
                ACE_Guard<ACE_Thread_Mutex> g(m_pingMutex);

                m_pingCondition.broadcast();
        }
        //
        // Otherwise, assume the command is solicited. Route it to the
        // first pending reader.
        else 
        {
                ACE_Guard<ACE_Thread_Mutex> g(m_pendingReplyQueue.GetMutex());

                if (m_pendingReplyQueue.IsEmpty_i())
                {
                        ACE_ERROR((LM_WARNING, "Got spurious message: %s\n", m_buf));
                }
                else
                {
                        std::string buf(m_buf);
                        m_pendingReplyQueue.SetReply_i(buf);
                }
        }

        return 0;
}



int AddressClient::SendCommand(const char *str, std::string &reply)
{
        //
        // A place to hold the result of method calls.
        int status = -1;

        {
                ACE_Guard<ACE_Thread_Mutex> g(m_pendingReplyQueue.GetMutex());

                //
                // Try to send the command over the network.
                status = this->m_socket.send(str, strlen(str), m_addrServer);
                if (status == -1)
                {
                        ACE_ERROR_RETURN((LM_ERROR, "(%N:%l) %p\n","send"), -1);
                }       
                
                //
                // If we have no timeout, wait indefinitely for the reply to arrive.
                if (m_timeout == ACE_Time_Value::zero)
                {
                        status = m_pendingReplyQueue.GetReply_i(reply);
                }
                //
                // Otherwise, wait until either the reply arrives or the timeout
                // expires.
                else
                {
                        ACE_Time_Value deadline = ACE_OS::gettimeofday() + m_timeout;
                        status = m_pendingReplyQueue.GetReply_i(reply, &deadline);
                }
        }

        //
        // If we failed to get the semaphore for any reason, log the error
        // and return failure.
        if (status == -1)
        {
                //
                // If the command failed due to timeout, log a more descriptive
                // error.  Other errors get the system-defined error message.
                if (ACE_OS::last_error() == ETIME)
                {
                        ACE_ERROR_RETURN((LM_ERROR,
                                "(%N:%l) Reply to command '%s' timed out.\n", str), -1);
                }
                else
                {
                        ACE_ERROR_RETURN((LM_ERROR, "(%N:%l) %p\n", "acquire"), -1);
                }
        }

        return 0;
}

---