RmsProducer.cpp

Go to the documentation of this file.

#include <rms/RmsProducer.h>
#include <rms/base/RmsRuntimeException.h>

namespace gov {

namespace fnal {

namespace cd {

namespace rms {

/**
 * Create a new producer given a connection and a destination.
 *
 * This constructor will not modify the reply timeout, which is set at
 * 5 minutes by default.
 *
 * @param providerConnection Connection object, which is used for receiving
 * and sending messages.
 * @param dest The destination that we are sending messages to.
 */
RmsProducer::RmsProducer(boost::shared_ptr<provider::RmsConnection> providerConnection, 
                         base::RmsDestination dest) :
    _providerConnection(providerConnection),
    _replyTimeout(300),
    _listenerLoop(&_inputQueue),
    _listenerThread(boost::bind(&novadaq::Runnable::run, &_listenerLoop)) {
    _myDestinations.push_back(dest);

    initialize();
}

/**
 * Create a new producer given a connection and a destination.
 *
 * This constructor will modify the reply timeout.
 *
 * @param providerConnection Connection object, which is used for receiving
 * and sending messages.
 * @param dest The destination that we are sending messages to.
 * @param replyTimeout The amount of time in seconds to wait for
 * a reply to a message.
 */
RmsProducer::RmsProducer(boost::shared_ptr<provider::RmsConnection> providerConnection, 
                         base::RmsDestination dest, int replyTimeout) :
    _providerConnection(providerConnection),
    _listenerLoop(&_inputQueue),
    _listenerThread(boost::bind(&novadaq::Runnable::run, &_listenerLoop)) {
    _replyTimeout = replyTimeout;
    
    _myDestinations.push_back(dest);
    
    initialize();
}

/**
 * Create a new producer given a connection and an array of destinations.
 *
 * This constructor will not modify the reply timeout, which is set at
 * 5 minutes by default.
 *
 * @param providerConnection Connection object, which is used for receiving
 * and sending messages.
 * @param dest The destinations that we are sending messages to.
 */
RmsProducer::RmsProducer(boost::shared_ptr<provider::RmsConnection> providerConnection, 
                         std::vector <base::RmsDestination> dest) :
    _providerConnection(providerConnection),
    _replyTimeout(300),
    _listenerLoop(&_inputQueue),
    _listenerThread(boost::bind(&novadaq::Runnable::run, &_listenerLoop)) {
    _myDestinations = dest;

    initialize();
}

/**
 * Create a new producer given a connection and an array of destinations.
 *
 * This constructor will modify the reply timeout.
 *
 * @param providerConnection Connection object, which is used for receiving
 * and sending messages.
 * @param dest The destinations that we are sending messages to.
 * @param replyTimeout The amount of time in seconds to wait for
 * a reply to a message.
 */
RmsProducer::RmsProducer(boost::shared_ptr<provider::RmsConnection> providerConnection, 
                         std::vector <base::RmsDestination> dest, int replyTimeout) :
    _providerConnection(providerConnection),
    _listenerLoop(&_inputQueue),
    _listenerThread(boost::bind(&novadaq::Runnable::run, &_listenerLoop)) {
    _replyTimeout = replyTimeout;
    
    _myDestinations = dest;
    
    initialize();
}

/**
 * Destructor, have close() do all the cleanup.
 */
RmsProducer::~RmsProducer() {
    close();
}

/**
 * Send a string to the destinations that are associated with
 * this producer.  No record of this is kept, so it won't be
 * possible to reply to this string.
 *
 * @param messageText The string to send out.
 */
void RmsProducer::sendString(std::string messageText) {
    for (unsigned int i = 0; i < _myDestinations.size(); i++) {
        _providerConnection->sendString(_myDestinations[i], messageText);
    }
}

/**
 * Send a message.  This is responsible for setting the replyDestination
 * of the message, the sentTime and the destination.  This will also
 * add the message's ID to the reply list.
 *
 * @param message The message to be sent.
 */
void RmsProducer::sendMessage(base::RmsMessage message) {
    base::RmsDestination replyDestination;
    const std::string messageUUID = _providerConnection->getUUID();
    boost::posix_time::ptime now(boost::posix_time::microsec_clock::universal_time());

    message.setSentTime();
    message.setId(messageUUID);

    replyDestination = _providerConnection->getMessageSource();

    // 21-Jun-2007, KAB - provide a default time to live (where should we get
    // this from officially?)
    message.setTimeToLive(5000);
  
    _replyIdTable.insert(std::pair<std::string, 
                         boost::posix_time::ptime>(messageUUID, now));

    for (unsigned int i = 0; i < _myDestinations.size(); i++) {
        message.setDestination(_myDestinations[i]);

        replyDestination.setProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME,
                                     _myDestinations[i].getProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME));
        replyDestination.validate();
        message.setReplyDestination(replyDestination);

        _providerConnection->sendMessage(_myDestinations[i], message);
    }
}

/**
 * Pull replies off the incoming message queue.  This function will
 * block until a message is received.
 *
 * @return An RmsMessage on success.  This will create an RmsStatus
 * object and insert it into the RmsMessage if an error occurs.
 */
boost::shared_ptr<base::RmsMessage> RmsProducer::receiveReply() {
    boost::shared_ptr<base::RmsMessage> receivedReply;
    boost::shared_ptr<std::string> messageText;

    do {
        messageText = _inputQueue.take();
        receivedReply.reset(new base::RmsMessage);
        receivedReply->deserialize(*messageText);
    } while (!verify(receivedReply));
  
    return receivedReply;
}

/**
 * Pull replies off the incoming message queue.  This function will
 * block until a message is received, or the time expires; whichever
 * happens first.
 *
 * @param timeout The maximum amount of time in seconds to wait if
 * the message queue is empty before giving up and returning.
 *
 * @return An RmsMessage on success.  This will return null if we fail
 * to pull a message off the queue.  If there is some sort of error in
 * pulling a message off the queue or in decoding the message, an RmsMessage
 * will be returned with an RmsStatus object inside it.
 */
boost::shared_ptr<base::RmsMessage> RmsProducer::receiveReply(long timeout) {
    boost::shared_ptr<base::RmsMessage> receivedReply;
    boost::shared_ptr<std::string> messageText;

    do {
        messageText = _inputQueue.poll(timeout);
        
        if (!messageText) {
            receivedReply.reset();
            return receivedReply;
        }

        receivedReply.reset(new base::RmsMessage);
        receivedReply->deserialize(*messageText);
    } while (!verify(receivedReply));
  
    return receivedReply;
}

/**
 * Pull replies off the incoming message queue.  This function will
 * not block.
 *
 * @return An RmsMessage on success.  This will return null if we timeout
 * while trying to pull a message off the queue.  If there is some sort of 
 * error in pulling a message off the queue or in decoding the message, an 
 * RmsMessage will be returned with an RmsStatus object inside it.
 */
boost::shared_ptr<base::RmsMessage> RmsProducer::receiveReplyNoWait() {
    boost::shared_ptr<base::RmsMessage> receivedReply;
    boost::shared_ptr<std::string> messageText;

    do {
        messageText = _inputQueue.poll();

        if (!messageText) {
            receivedReply.reset();
            return receivedReply;
        }

        receivedReply.reset(new base::RmsMessage);
        receivedReply->deserialize(*messageText);
    } while (!verify(receivedReply));

    return receivedReply;
}

/**
 * Register a callback with the producer.  This will start a
 * thread that pulls replies off the input queue and sends them
 * to the listener.
 *
 * @param listener The listener that will receive messages
 *
 * @return NONE
 */
void RmsProducer::setReplyListener(RmsMessageListener *listener) {
    _listenerLoop.setMessageListener(listener);
}

/**
 * Stop the delete the listener loop, as well as the
 * provider listener and input queue.
 */
void RmsProducer::close() {
    _listenerLoop.stop();
    _listenerThread.join();

    if (_providerListener) {
        delete _providerListener;
        _providerListener = NULL;
    }

    return;
}

/**
 * Verify that a reply received is one that the producer
 * is actually interested in.  We maintain a list of all
 * the message IDs that are sent out, and compare replies
 * received to this list.
 *
 * @param message The message that we are checking.
 *
 * @return True is the message is supposed to come to
 * this producer, false otherwise.
 */
bool RmsProducer::verify(boost::shared_ptr<base::RmsMessage> message) {
    boost::posix_time::ptime now(boost::posix_time::microsec_clock::universal_time());
    boost::posix_time::ptime deadline = now - boost::posix_time::seconds(_replyTimeout);
    std::map<std::string, boost::posix_time::ptime>::iterator replyIterator;

    if(message->getMessageStatus().getCode() != base::RmsStatus::SUCCESS_CODE) {
        return true;
    }

    /**
     * Look through the list of IDs so we can nuke any that have
     * expired.
     */
    replyIterator = _replyIdTable.begin();
    
    while (replyIterator != _replyIdTable.end()) {
        if (replyIterator->second < deadline) {
            _replyIdTable.erase(replyIterator);
        }
        replyIterator++;
    }

    /**
     * Try to locate the ID of the message that was passed into
     * this method.
     */
    replyIterator = _replyIdTable.find(message->getCorrelationId());

    if (replyIterator != _replyIdTable.end()) {
        _replyIdTable.erase(message->getCorrelationId());
        return true;
    }

    return false;
}

/**
 * Called by the constructor, this will verify that our provider
 * can talk to the supplied destination, as well as creating the
 * input queue, replyId list, the provider listener.  This will
 * also determine the reply destination.
 */
void RmsProducer::initialize() {
    unsigned int i;
    base::RmsDestination replyDestination(_providerConnection->getMessageSource());

    for (i = 0; i < _myDestinations.size(); i++) {
        _replyChannels.insert(_myDestinations[i].getProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME));

        _myDestinations[i].validate();
        if (!_providerConnection->supportsDestination(_myDestinations[i])) {
            std::string msg;
            msg.append("The current connection does not support ");
            msg.append("the requested destination: ");
            msg.append(_myDestinations[i].toString());
            GENERATE_RMS_RUNTIME_EXCEPTION(msg);
        }
    }

    _listenerLoop.setMessageFilter(this);

    for (std::set <std::string>::iterator channelIterator = _replyChannels.begin();
         channelIterator != _replyChannels.end(); channelIterator++) {
        replyDestination.setProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME,
                                     *channelIterator);
        replyDestination.validate();

        _providerListener = new provider::BufferedProviderListener(&_inputQueue);
        _providerConnection->addListener(replyDestination, _providerListener);
    }
}

}  // end of namespace rms

}  // end of namespace cd

}  // end of namespace fnal

}  // end of namespace gov