ClientListenerLoop.cpp

Go to the documentation of this file.

#include <rms/ClientListenerLoop.h>
#include <rms/base/RmsMessage.h>
#include <xsd/cxx/tree/exceptions.hxx>

namespace gov {
    
namespace fnal {
        
namespace cd {

namespace rms {

/**
 * Create a new ClientListenerLoop with not client listner and no 
 * message filter.  The loop will just spin until a client listener
 * has been set.
 *
 * @param inputQueue The input queue to pull messages off of.
 */
ClientListenerLoop::ClientListenerLoop(util::LinkedBlockingQueue <boost::shared_ptr<std::string> > *inputQueue) {
    _inputQueue = inputQueue;
    _clientListener = NULL;
    _messageFilter = NULL;
}

/**
 * Destructor for the ClientListenerLoop.  This sets the
 * stop flag to true so that the thread will exit.  The
 * proper way to shutdown the loop is to call the stop()
 * method and then join() on the thread handle.  Otherwise,
 * its possible the thread could exit while holding the input
 * queue lock which may cause a crash.
 */
ClientListenerLoop::~ClientListenerLoop() {
    _stopRequested = true;

    while(_running) {
        usleep(100000);
    }
}

/**
 * Set the stop flag so that the thread exits.  After doing this,
 * join() on the thread handle to make sure that the thread
 * has exited.
 */
void ClientListenerLoop::stop() {
    _stopRequested = true;
}

/**
 * This method runs in its own thread.  If no client listener has been
 * set, it will spin until one is set.  After that, it will pull messages
 * off of the queue, deserialize them, and then pass them through a
 * message filter if one has been set.  Lastly, the message is passed
 * to the listener.
 */
void ClientListenerLoop::run() {
    boost::shared_ptr<std::string> messageText;

    _stopRequested = false;
    _running = true;

    while (!_stopRequested) {
        if (_clientListener) {
            messageText = _inputQueue->poll(1);
        }
        else {
            usleep(200000);
            continue;
        }
            
        if (!messageText) {
            continue;
        }
        // 19-Jun-2007, KAB - we need to create a new RmsMessage each time
        // since XSD deserialization doesn't overwrite existing element 
        // objects.
        boost::shared_ptr<base::RmsMessage> 
            receivedMessage (new base::RmsMessage);
        
        try {
            receivedMessage->deserialize(*messageText);
        }
        catch (const ::xsd::cxx::tree::exception<char> &excpt) {
            std::cout << "Exception thrown during deserialization ";
            std::cout << "of RMS message!" << std::endl;
            std::cout << excpt << std::endl;
        }
        
        if (_messageFilter) {
            if (_messageFilter->verify(receivedMessage)) {
                _clientListener->messageReceived(receivedMessage);
            }
        }
        else {
            _clientListener->messageReceived(receivedMessage);
        }
    }

    _running = false;
}

/**
 * Set the message listener that gets called when a message
 * is received.
 *
 * @param listener The listener that will get called when
 * a message is received.
 */
void ClientListenerLoop::setMessageListener(RmsMessageListener *listener) {
    _clientListener = listener;
}

/**
 * Set the message filter that will get called everytime
 * a message is pulled off the queue to verify that it
 * is relevant to the listener.
 *
 * @param filter The message filter
 */
void ClientListenerLoop::setMessageFilter(MessageFilter *filter) {
    _messageFilter = filter;
}

}  // end of namespace rms

}  // end of namespace cd

}  // end of namespace fnal

}  // end of namespace gov