common/messageable.cc

Go to the documentation of this file.

//

#include "consmgr.h"
#include "messageable.h"
#include "connlist.h"
#include "guard.h"

//
// Construction/destruction
//
Messageable::Messageable(void)
{
    int         ret;

    ret = pthread_mutex_init(&msg_mutex, NULL);
    PTHREAD_CHECK_AND_THROW(ret, "mutex_init(msg_mutex)");
    ret = pthread_cond_init(&msg_avail, NULL);
    PTHREAD_CHECK_AND_THROW(ret, "mutex_init(msg_mutex)");
}

Messageable::~Messageable(void)
{
    // I don't think we should care if this fails...
    (void)pthread_mutex_destroy(&msg_mutex);
    (void)pthread_cond_destroy(&msg_avail);

    // XXX Where do I look to know how to take myself out of any ConnectionList's that may know about me?
#warning "Messageable wants to know how to remove itself from everyone elses lists, but doesn't know how to!"
#if SHOULD_NOT_BE_HERE
    // And, make sure no one else will try to talk to us...
    ConnectionList::outlet_list.mutating_for_each(invalidate_messageable(this));
#endif
}


//
// Functionality methods
//

void
Messageable::queue_message(MessageP &msg)
{
    int         ret;
    clog << pthread_self() << " Locking Messageable's mutex " << &msg_mutex << endl;
    Guard       locker(&msg_mutex);

    clog << pthread_self() << " Locked " << &msg_mutex << ", pushing message onto the queue" << endl;
    msg_queue.push_back(msg);

    clog << "Messageable::queue_message() (" << this->get_id() << ") accepting a message and broadcasting on the condition variable" << endl;
    ret = pthread_cond_broadcast(&msg_avail);
    PTHREAD_CHECK_AND_THROW(ret, "signal_broadcast(msg_avail)");

    clog << pthread_self() << " returning from queue_message" << endl;
    return;
}

MessageP
Messageable::get_message(void)
{
    int         ret;
    clog << pthread_self() << " Gonna lock the message mutex " << &msg_mutex << endl;
    Guard       locker(&msg_mutex);

    /* We've locked the mutex.  Check to make sure there isn't data in */
    /* the queue already.  If there isn't, wait until there is... */
    while (msg_queue.empty()) {
        clog << pthread_self() << " Waiting on msg condition variable" << endl;
        ret = pthread_cond_wait(&msg_avail, &msg_mutex);
        PTHREAD_CHECK_AND_THROW(ret, "cond_wait(msg_avail, msg_mutex)");
        /* If someone else beat us to the data, just wait again... */
    }

    MessageP    retmsg = msg_queue.front();
    msg_queue.pop_front();

    return (retmsg);
}

// Handle a message passed to us.  Return a Blob* of data
// if we have data to be written (to ourself), or NULL to
// indicate that no data is to be written.  Anything else
// needs to be taken care of internally.
Data *
Messageable::handle_message(MessageP &msg)
{
    int         ret;

    clog << "I (" << this << ") got a message!  Type " << msg->name() << endl;

    // Handle certain simple commands...
    if (msg->type() == Message::M_Command) {
        switch (msg->code()) {
          case Message::C_Exit:
            clog << "Got an Exit command, need to shut down..." << endl;
#if 0
            if ((ret = pthread_mutex_trylock(&mutex)) == 0) {
                pthread_mutex_unlock(&mutex);
                MUTEX_LOG("Calling shutdown, mutex (" << &mutex << ") unlocked");
            } else {
                if (ret == EBUSY)
                    MUTEX_LOG("Calling shutdown, but mutex (" << &mutex << ") is locked already.");
                else
                    MUTEX_LOG("Error!  trylock of mutex (" << &mutex << ") failed: " << strerror(ret));
            }
#endif
            {
            Connection *c = dynamic_cast<Connection*>(this);
            // XXX - What do we do to shut down non-Connections?
            c->shutdown();
            }
            break;
          default:
            clog << "Got a command (" << msg->code() << "), ignoring...\n" << endl;
            break;
        }
    } else {
        /* msg->type() == Message::M_Data */
        clog << "msg->type() is " << msg->name() << endl;
        Data *d = dynamic_cast<Data*>(MessageP::GetPointer(msg));
        if (d == NULL) {
            throw(fireball(114,"Ahh!  Can't get a Data* from the msg in Messageable::handle_message"));
        }
        /* A new MessageP will be constructed with this Data, so it needs */
        /* to be dynamically allocated. */
        return(new Data(*d));
    }

    return(NULL);
}

---