RmsReceiver.h

Go to the documentation of this file.

#ifndef _RMS_RECEIVER_H
#define _RMS_RECEIVER_H

/**
 * @package gov.fnal.cd.rms
 *
 * @brief Classes that allow someone to send and receive
 * messages utilizing the Responsive Messaging System.
 *
 * This namespace includes the Sender and Receiver classes
 * as well as interfaces that can be implemented to handle
 * messages straight out of RMS.
 */

#include <NovaDAQUtilities/Runnable.h>
#include <rms/base/RmsCloseable.h>
#include <rms/base/RmsDestination.h>
#include <rms/base/RmsRuntimeException.h>

#include <boost/shared_ptr.hpp>
#include <boost/thread/thread.hpp>
#include <boost/interprocess/smart_ptr/unique_ptr.hpp>

#include <string>
#include <iostream>

namespace gov {

namespace fnal {

namespace cd {

namespace rms {

template<typename R> struct RunnableDeleter {
    void operator()(R *runnable) {
        assert( runnable != NULL );
        runnable->stop();
        delete runnable;
    }
};

template<typename T> struct BoostThreadDeleter {
    void operator()(T * thread) {
        assert( thread != NULL );

//        thread->interrupt();

        if (thread->joinable())
            thread->join();

        delete thread;
    }
};


template <typename MSG>
class RmsDummyListener {
public:
    void processMessage(boost::shared_ptr<MSG> message);
};

template <typename MSG>
void RmsDummyListener<MSG>::processMessage(boost::shared_ptr<MSG> message)
{
    // do nothing
}


template <class CONN, typename MSG, class LSTNR>
class ListenerLoop : public novadaq::Runnable {
public:
    ListenerLoop(boost::shared_ptr<CONN> connection,
                 unsigned int receiverHandle,
                 boost::shared_ptr<LSTNR> listener,
                 std::string targetName);
    ListenerLoop(boost::shared_ptr<CONN> connection,
                 unsigned int receiverHandle,
                 LSTNR* listener,
                 std::string targetName);
    void run();

private:
    boost::shared_ptr<CONN> _myConnection;
    unsigned int _receiverHandle;
    boost::shared_ptr<LSTNR> _listener;
    LSTNR* _listenerRawPtr;
    std::string _targetName;
};

template <class CONN, typename MSG, class LSTNR>
ListenerLoop<CONN, MSG, LSTNR>::
ListenerLoop(boost::shared_ptr<CONN> connection,
             unsigned int receiverHandle,
             boost::shared_ptr<LSTNR> listener,
             std::string targetName)
{
    _myConnection = connection;
    _receiverHandle = receiverHandle;
    _listener = listener;
    _listenerRawPtr = 0;
    _targetName = targetName;
}

template <class CONN, typename MSG, class LSTNR>
ListenerLoop<CONN, MSG, LSTNR>::
ListenerLoop(boost::shared_ptr<CONN> connection,
             unsigned int receiverHandle,
             LSTNR* listener,
             std::string targetName)
{
    _myConnection = connection;
    _receiverHandle = receiverHandle;
    _listenerRawPtr = listener;
    _targetName = targetName;
}

template <class CONN, typename MSG, class LSTNR>
void ListenerLoop<CONN, MSG, LSTNR>::run() {
    _stopRequested = false;
    _running = true;

    while (! _stopRequested) {
        try {
            if (_listener.get() != 0) {
                boost::shared_ptr<MSG> message(new MSG());
                //std::cout << "***** Before thread receive " << _targetName
                //          << " " << _receiverHandle << std::endl;
                if (_myConnection->receiveMessage(_receiverHandle, *message,
                                                  250000)) {
                    _listener->processMessage(message);
                }
            }
            else if (_listenerRawPtr != 0) {
                boost::shared_ptr<MSG> message(new MSG());
                //std::cout << "***** Before thread receive " << _targetName
                //          << " " << _receiverHandle << std::endl;
                if (_myConnection->receiveMessage(_receiverHandle, *message,
                                                  250000)) {
                    _listenerRawPtr->processMessage(message);
                }
            }
            else {
                //interruptable sleep
                boost::this_thread::sleep(boost::posix_time::microseconds(250000));
            }
        }
        catch ( gov::fnal::cd::rms::base::RmsNotConnectedException& ex  ) {
            LOG_EXCEPTION("Caught RmsNotConnectedException in ListenerLoop::run(); message=", ex);

            //interruptable sleep
            boost::this_thread::sleep(boost::posix_time::microseconds(250000));
        }
        catch (gov::fnal::cd::rms::base:: RmsExitingProcessException& ex1) {
            _myConnection->close();   //does not throw

            _stopRequested=true;

            LOG_EXCEPTION("Caught RmsNotConnectedException in ListenerLoop::run(); message=", ex1);
        }
        catch ( ... ) {
            _myConnection->close();   //does not throw
          
            _stopRequested=true;

            LOG_EXCEPTION("Caught RmsNotConnectedException in ListenerLoop::run(); message=","");
        }
    }

    _running = false;
    
    #if CETDDS_DEBUG1
    std::cout << "Exiting an RmsReceiver listener loop." << std::endl;
    #endif
}



/**
 * This class contains functionality for receiving messages of specified
 * types from a specified destination.
 *
 * @author Kurt Biery
 * @version $Revision: 1.12 $ $Date: 2011/05/12 17:11:14 $
 */

template <class CONN, typename MSG, class LSTNR = RmsDummyListener<MSG> >
class RmsReceiver : public base::RmsCloseable {

public:
    RmsReceiver(boost::shared_ptr<CONN> parentConnection,
                base::RmsDestination dest);
    ~RmsReceiver();

    void receiveMessage(MSG& message);

    bool receiveMessage(MSG& message, unsigned int usecTimeout);

    void setListener(boost::shared_ptr< LSTNR > listener);

    void setListener(LSTNR* listener);

    void close() throw();
    bool hasBeenClosed() throw(){return _hasBeenClosed;};
private:
    bool isRunsInListenerLoop();

private:
    /**
     * Handle to the RMS connection
     */
    boost::shared_ptr<CONN> _myConnection;

    /**
     * RmsDestination that we are listening to
     */
    base::RmsDestination _myDestination;

    unsigned int _receiverHandle;

    boost::interprocess::unique_ptr< ListenerLoop<CONN, MSG, LSTNR>,
          RunnableDeleter< ListenerLoop<CONN, MSG, LSTNR> > > _listenerLoop;

          
    boost::interprocess::unique_ptr< boost::thread,
          BoostThreadDeleter<boost::thread> > _listenerThread;

    bool _hasBeenClosed;
};

/**
 * Constructs an RmsReceiver instance.
 *
 * @param parentConn Connection object which is used for
 * receiving messages.
 * @param dest The destination that we are receiving messages from.
 * @throws RmsRuntimeException If the destination is invalid.
 */
template <class CONN, typename MSG, class LSTNR>
RmsReceiver<CONN, MSG, LSTNR>::
RmsReceiver(boost::shared_ptr<CONN> parentConn, base::RmsDestination dest) :
        _myConnection(parentConn),
        _myDestination(dest),
        _listenerLoop(0),
        _listenerThread(0),
        _hasBeenClosed(false)
{
    dest.validate();
    if (! _myConnection->supportsDestination(dest)) {
        std::string msg;
        msg.append("The current connection does not support ");
        msg.append("the requested destination: ");
        msg.append(dest.toString());
        GENERATE_RMS_RUNTIME_EXCEPTION(msg);
    }

    MSG dummyObj;
    _receiverHandle = _myConnection->registerReceiver(dummyObj, _myDestination);
}

/**
 * Destructs an RmsReceiver instance.
 */
template <class CONN, typename MSG, class LSTNR>
RmsReceiver<CONN, MSG, LSTNR>::~RmsReceiver()
{
    this->close();  //does not throw
}

/**
 * Checks if an RmsReceiver instance runs in a ListenerLoop.
 */
template <class CONN, typename MSG, class LSTNR>
bool RmsReceiver<CONN, MSG, LSTNR>::isRunsInListenerLoop()
{
    return (_listenerLoop.get() != 0 && _listenerThread.get() != 0);
}


/**
 * Receives the next message.  This function will block until a
 * message is received.
 *
 * @param the message reference to copy the received message into
 * @throws RmsCommunicationException if there is a problem receiving
 *         the next message.
 */
template <class CONN, typename MSG, class LSTNR>
void RmsReceiver<CONN, MSG, LSTNR>::receiveMessage(MSG& message)
{
    try
    {
        if (_hasBeenClosed)
            GENERATE_RMS_NOTCONNECTED_EXCEPTION("RMS connection was closed.");

        _myConnection->receiveMessage(_receiverHandle, message);
    }
    catch ( gov::fnal::cd::rms::base::RmsNotConnectedException& ex  ) {
        LOG_EXCEPTION(  "Caught RmsNotConnectedException in RmsReceiver::receiveMessage(); message=", ex );

        //rethrow if running inside a listener thread
        if (isRunsInListenerLoop())
            throw;
    }
    catch ( ... ) {
        this->close();  //does not throw

        LOG_EXCEPTION( "Caught exception in RmsReceiver::receiveMessage(). Closed RMS connection.","");

        //rethrow if running inside a listener thread
        if (isRunsInListenerLoop())
            throw;
    }
}

/**
 * Receives the next message.  This function will return after the
 * specified timeout if no message has been received in that time.
 *
 * @param the message reference to copy the received message into
 * @param the amount of time (in microseconds) to wait for a message
 * @returns true if a message was received, false otherwise
 * @throws RmsCommunicationException if there is a problem receiving
 *         the next message.
 */
template <class CONN, typename MSG, class LSTNR>
bool RmsReceiver<CONN, MSG, LSTNR>::receiveMessage(MSG& message,
        unsigned int usecTimeout)
{
    try
    {
        if (_hasBeenClosed)
            GENERATE_RMS_NOTCONNECTED_EXCEPTION("RMS connection was closed.");

        return _myConnection->receiveMessage(_receiverHandle, message,
                                             usecTimeout);
    }
    catch ( gov::fnal::cd::rms::base::RmsNotConnectedException& ex  ) {
        LOG_EXCEPTION("Caught RmsNotConnectedException in RmsReceiver::receiveMessage(); message=", ex);

        //rethrow if running inside a listener thread
        if (isRunsInListenerLoop())
            throw;
    }
    catch ( ... ) {
        this->close();  //does not throw

        LOG_EXCEPTION("Caught exception in RmsReceiver::receiveMessage(). Closing RMS connection.","");

        //rethrow if running inside a listener thread
        if (isRunsInListenerLoop())
            throw;
    }

    return false;
}

/**
 * Sets the listener for this receiver.  *Note* that this method starts
 * a background thread to receive messages, *and* the presence of a
 * listener makes the receiving of messages with the
 * receiveMessage() method rather irrelevant.
 *
 * @param the listener to be called when messages arrive
 */
template <class CONN, typename MSG, class LSTNR>
void RmsReceiver<CONN, MSG, LSTNR>::
setListener(boost::shared_ptr<LSTNR> listener)
{
    // clean up any existing listener
    if (_listenerLoop.get() != 0) {
        _listenerLoop->stop();
    }

    _listenerThread.reset(); //joins and deletes the listener thread

    std::string targetProperty =
        _myDestination.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME);

    _listenerLoop.reset(
        new ListenerLoop<CONN, MSG, LSTNR>(_myConnection, _receiverHandle,
                                           listener, targetProperty));

    _listenerThread.reset(
        new boost::thread(boost::bind(&novadaq::Runnable::run, _listenerLoop.get())));
}

/**
 * Sets the listener for this receiver.  *Note* that this method starts
 * a background thread to receive messages, *and* the presence of a
 * listener makes the receiving of messages with the
 * receiveMessage() method rather irrelevant.
 *
 * @param the listener to be called when messages arrive
 */
template <class CONN, typename MSG, class LSTNR>
void RmsReceiver<CONN, MSG, LSTNR>::
setListener(LSTNR* listener)
{
    // clean up any existing listener
    if (_listenerLoop.get() != 0) {
        _listenerLoop->stop();
    }

    _listenerThread.reset(); //joins and deletes the listener thread

    std::string targetProperty =
        _myDestination.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME);

    _listenerLoop.reset(
        new ListenerLoop<CONN, MSG, LSTNR>(_myConnection, _receiverHandle,
                                           listener, targetProperty));

    _listenerThread.reset(new boost::thread(
        boost::bind(&novadaq::Runnable::run, _listenerLoop.get())));
}

/**
 * Closes the RmsReceiver.
 */
template <class CONN, typename MSG, class LSTNR>
void RmsReceiver<CONN, MSG, LSTNR>::close() throw()
{
    if (_hasBeenClosed) {
        return;
    }

    _hasBeenClosed = true;

    if (_listenerLoop.get() != 0) {
        _listenerLoop->stop();    //not expected to throw
    }

    _listenerThread.reset(); //joins and deletes the listener thread
    _listenerLoop.reset(); //deletes the listener loop

    try {
        _myConnection->unRegisterReceiver(_receiverHandle);
    } catch (...) {}

    _myConnection.reset();
#if CETDDS_DEBUG1
    std::cout << "Closed RmsReceiver connection" << std::endl;
#endif
}

};  // end of namespace rms

};  // end of namespace cd

};  // end of namespace fnal

};  // end of namespace gov

#endif
// kate: indent-mode cstyle; space-indent on; indent-width 0;