DDSConnection.h

Go to the documentation of this file.

#ifndef _RMS_DDSCONNECTION_H
#define _RMS_DDSCONNECTION_H

#include <rms/base/RmsDestination.h>
#include <rms/util/UUIDGenerator.h>
#include <rms/provider/CETDDS.h>
#include <NovaDAQUtilities/TimeUtils.h>

#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/algorithm/string.hpp>

#include <string>
#include <map>

namespace gov {

namespace fnal {

namespace cd {

namespace rms {

namespace provider {

/**
 * This class takes care of setting up and managing connections
 * to DDS for RMS.
 *
 * @author Kurt Biery
 * @version $Revision: 1.17.14.1 $ $Date: 2019/09/27 00:07:31 $
 */

class DDSConnection {

 public:

    DDSConnection (const std::string applicationName,
                   const int partitionNumber);
    ~DDSConnection ();

    bool supportsDestination(const base::RmsDestination& candidateDestination);

    std::string getApplicationName() const {return _applicationName;}
    int getPartitionNumber() const {return _partitionNumber;}

    void close();

    /**
     * Get a new human readable UUID from the UUID generator.
     *
     * @return A new UUID.
     */
    const std::string getUUID() {
        return _uuidGenerator->getUUIDHex();
    }

    std::string getHostName() {
        return _myHostName;
    }

    template <class T>
    void dumpMessage(T const& message, bool isIncomingMessage) {
        std::cout << "++++++++++++++++++++++++++++++++++++++++" << std::endl;
        std::cout << "Message type = " << typeid(message).name() << std::endl;
        std::cout << "ID = " << message.header.id << std::endl;
        std::cout << "Target = " << message.header.target << std::endl;
        std::cout << "Channel = " << message.header.channel << std::endl;
        std::cout << "Reply target = " << message.header.replyTarget << std::endl;
        std::cout << "Correlation ID = " << message.header.correlationId << std::endl;
        std::cout << "Sent time = " << message.header.sentTime.sec
                  << " " << message.header.sentTime.usec << std::endl;
        std::cout << "Correlation time = " << message.header.correlationTime.sec
                  << " " << message.header.correlationTime.usec << std::endl;
        std::cout << "Priority = " << message.header.priority << std::endl;
        std::cout << "Partition number = " << message.header.partitionNumber << std::endl;
        std::cout << "Host name = " << message.header.hostName << std::endl;
        std::cout << "Application name = " << message.header.applicationName << std::endl;

        if (isIncomingMessage && message.header.correlationTime.sec > 1000000) {
            struct timeval then;
            then.tv_sec = message.header.correlationTime.sec;
            then.tv_usec = message.header.correlationTime.usec;
            struct timeval now;
            gettimeofday(&now, 0);

            double thenTime = novadaq::convertTimevalToDouble(then);
            double nowTime = novadaq::convertTimevalToDouble(now);
            double timeDiff = 1000.0 * (nowTime - thenTime);

            std::cout << "Request/reply message time = " << timeDiff << " msec" << std::endl;
        }

        std::cout << "----------------------------------------" << std::endl;
    }

    /**
     * Sends a message of the specified type to the specified destination.
     *
     * @param dest The destination to send the message to.
     * @param message The message to be sent to the destination.
     */
    template <class T>
    void sendMessage(unsigned int senderHandle, T& message) {
        std::map<unsigned int,
            boost::shared_ptr<base::RmsCloseable> >::const_iterator writerIter;
        {
            boost::mutex::scoped_lock sl(_registrationMutex);
            writerIter = _writerTable.find(senderHandle);
        }

        if (writerIter != _writerTable.end()) {

            std::string id = this->getUUID();
            message.header.id = DDS::string_dup(id.c_str());

            std::map<unsigned int,
                std::pair<std::string, std::string> >::const_iterator tgtChanIter;
            tgtChanIter = _targetChannelTable.find(senderHandle);
            if (tgtChanIter != _targetChannelTable.end()) {
                std::string target = tgtChanIter->second.first;
                std::string channel = tgtChanIter->second.second;
                message.header.target = DDS::string_dup(target.c_str());
                message.header.channel = DDS::string_dup(channel.c_str());
            }

            std::string replyTarget = this->getApplicationName();
            message.header.replyTarget = DDS::string_dup(replyTarget.c_str());
            message.header.applicationName = DDS::string_dup(replyTarget.c_str());

            struct timeval now;
            gettimeofday(&now, 0);
            message.header.sentTime.sec = now.tv_sec;
            message.header.sentTime.usec = now.tv_usec;

            message.header.partitionNumber = this->getPartitionNumber();

            message.header.hostName = DDS::string_dup(_myHostName.c_str());

            if ((_debugFlag & 0x01) != 0) {
                dumpMessage(message, false);
            }

            base::RmsCloseable *obj = writerIter->second.get();
            DDSTopicWriterClass<T> *myWriter =
                dynamic_cast< DDSTopicWriterClass<T>* >(obj);
            myWriter->write(message);
        }
    }

    /**
     * Receives a message of the specified type from the specified destination.
     *
     * @param dest The destination to send the message to.
     * @param message The message received from the destination.
     */
    template <class T>
    void receiveMessage(unsigned int receiverHandle, T& message) {
        std::map<unsigned int,
            boost::shared_ptr<base::RmsCloseable> >::const_iterator readerIter;
        {
            boost::mutex::scoped_lock sl(_registrationMutex);
            readerIter = _readerTable.find(receiverHandle);
        }

        if (readerIter != _readerTable.end()) {
            base::RmsCloseable *obj = readerIter->second.get();
            DDSTopicReaderClass<T> *myReader =
                dynamic_cast< DDSTopicReaderClass<T>* >(obj);
            myReader->take_wait(message);
        }
        
        if ((_debugFlag & 0x01) != 0) {
            dumpMessage(message, true);
        }
    }

    /**
     * Waits for a message of the specified type to arrive from the
     * specified destination, but returns after the specified timeout
     * if no message has arrived.
     *
     * @param dest The destination to send the message to.
     * @param message The message received from the destination.
     * @param usecTimeout Amount of time to wait for a message (microseconds).
     * @return true if a message was received, false if not.
     */
    template <class T>
    bool receiveMessage(unsigned int receiverHandle, T& message,
                        unsigned int usecTimeout) {
        std::map<unsigned int,
            boost::shared_ptr<base::RmsCloseable> >::const_iterator readerIter;
        {
            boost::mutex::scoped_lock sl(_registrationMutex);
            readerIter = _readerTable.find(receiverHandle);
        }

        if (readerIter != _readerTable.end()) {
            base::RmsCloseable *obj = readerIter->second.get();
            DDSTopicReaderClass<T> *myReader =
                dynamic_cast< DDSTopicReaderClass<T>* >(obj);
            DDS::ReturnCode_t status =
                myReader->take_timed_wait(message, usecTimeout);

            if ((_debugFlag & 0x01) != 0) {
                if (status == DDS::RETCODE_OK) {
                    dumpMessage(message, true);
                }
            }
            if ((_debugFlag & 0x02) != 0) {
                if (status == DDS::RETCODE_TIMEOUT) {
                    std::cout << "++++++++++++++++++++++++++++++++++++++++" << std::endl;
                    std::cout << "Timeout waiting for " << typeid(message).name()
                              << " message." << std::endl;
                    std::cout << "Waited " << usecTimeout << " microseconds." << std::endl;
                }
                else if (status != DDS::RETCODE_OK) {
                    std::cout << "++++++++++++++++++++++++++++++++++++++++" << std::endl;
                    std::cout << "Received error code " << status
                              << " from take_timed_wait call."<< std::endl;
                }
            }

            return (status == DDS::RETCODE_OK);
        }
        return false;
    }

    template <class T>
    unsigned int registerSender(T const& sampleMessage,
                                base::RmsDestination const& dest) {
        boost::mutex::scoped_lock sl(_registrationMutex);

        std::string targetProperty =
            dest.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME);
        std::string channelProperty =
            dest.getProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME);
        std::string topicName = targetProperty + "_" + channelProperty;
        boost::replace_all(topicName, "-", "_dash_");
        std::string partitionName = "Partition" +
            boost::lexical_cast<std::string>(_partitionNumber);

        boost::shared_ptr<base::RmsCloseable>
            myWriter(new DDSTopicWriterClass<T>(topicName.c_str(),
                                                partitionName.c_str()));
        ++_writerCounter;
        _writerTable[_writerCounter] = myWriter;
        _targetChannelTable[_writerCounter] =
            make_pair(targetProperty, channelProperty);
        return _writerCounter;
    }

    void unRegisterSender(unsigned int senderHandle) {
        boost::mutex::scoped_lock sl(_registrationMutex);

        std::map<unsigned int,
            boost::shared_ptr<base::RmsCloseable> >::const_iterator writerIter;
        writerIter = _writerTable.find(senderHandle);

        if (writerIter != _writerTable.end()) {
            _writerTable.erase(senderHandle);
        }

        std::map<unsigned int,
            std::pair<std::string, std::string> >::const_iterator tgtChanIter;
        tgtChanIter = _targetChannelTable.find(senderHandle);

        if (tgtChanIter != _targetChannelTable.end()) {
            _targetChannelTable.erase(senderHandle);
        }
    }

    template <class T>
    unsigned int registerReceiver(T const& sampleMessage,
                                  base::RmsDestination const& dest) {
        boost::mutex::scoped_lock sl(_registrationMutex);

        std::string targetProperty =
            dest.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME);
        std::string channelProperty =
            dest.getProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME);
        std::string topicName = targetProperty + "_" + channelProperty;
        boost::replace_all(topicName, "-", "_dash_");
        std::string partitionName = "Partition" +
            boost::lexical_cast<std::string>(_partitionNumber);

        boost::shared_ptr<base::RmsCloseable>
            myReader(new DDSTopicReaderClass<T>(topicName.c_str(),
                                                partitionName.c_str()));
        ++_readerCounter;
        _readerTable[_readerCounter] = myReader;
        return _readerCounter;
    }

    void unRegisterReceiver(unsigned int receiverHandle) {
        boost::mutex::scoped_lock sl(_registrationMutex);

        std::map<unsigned int,
            boost::shared_ptr<base::RmsCloseable> >::const_iterator readerIter;
        readerIter = _readerTable.find(receiverHandle);

        if (readerIter != _readerTable.end()) {
            _readerTable.erase(receiverHandle);
        }
    }

 protected:

 private:

    util::UUIDGenerator *_uuidGenerator;
    std::string _applicationName;
    int _partitionNumber;

    std::map<unsigned int, boost::shared_ptr<base::RmsCloseable> > _writerTable;
    std::map<unsigned int, std::pair<std::string, std::string> >
        _targetChannelTable;
    unsigned int _writerCounter;

    std::map<unsigned int, boost::shared_ptr<base::RmsCloseable> > _readerTable;
    unsigned int _readerCounter;

    std::string _myHostName;

    mutable boost::mutex _registrationMutex;

    int _debugFlag;

    static int _instanceCount;
    static boost::mutex _staticMutex;
};

}  // end of namespace provider

}  // end of namespace rms

}  // end of namespace cd

}  // end of namespace fnal

}  // end of namespace gov

#endif