#include "/cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N21-01-23/rms/cxx/include/provider/DDSConnection.h"

Public Member Functions
	DDSConnection (const std::string applicationName, const int partitionNumber)

	~DDSConnection ()

bool	supportsDestination (const base::RmsDestination &candidateDestination)

std::string	getApplicationName () const

int	getPartitionNumber () const

void	close ()

const std::string	getUUID ()

std::string	getHostName ()

template<class T >
void	dumpMessage (T const &message, bool isIncomingMessage)

template<class T >
void	sendMessage (unsigned int senderHandle, T &message)

template<class T >
void	receiveMessage (unsigned int receiverHandle, T &message)

template<class T >
bool	receiveMessage (unsigned int receiverHandle, T &message, unsigned int usecTimeout)

template<class T >
unsigned int	registerSender (T const &sampleMessage, base::RmsDestination const &dest)

void	unRegisterSender (unsigned int senderHandle)

template<class T >
unsigned int	registerReceiver (T const &sampleMessage, base::RmsDestination const &dest)

void	unRegisterReceiver (unsigned int receiverHandle)

Private Attributes
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

boost::mutex	_registrationMutex

int	_debugFlag

Static Private Attributes
static int	_instanceCount = 0

static boost::mutex	_staticMutex

Detailed Description

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

Definition at line 35 of file DDSConnection.h.

Constructor & Destructor Documentation

gov::fnal::cd::rms::provider::DDSConnection::DDSConnection	(	const std::string	applicationName,
		const int	partitionNumber
	)

Creates a new DDSConnection given the application name and partition.

Parameters

applicationName	The name of the application generating/receiving messages.
applicationPartition	The partition of the application that is generating/receiving messages.

Definition at line 27 of file DDSConnection.cpp.

References _applicationName, _debugFlag, _instanceCount, _myHostName, _partitionNumber, _staticMutex, _uuidGenerator, om::cerr, gov::fnal::cd::rms::provider::DPSingleton::connect_participant(), allTimeWatchdog::endl, cet::getenv(), gov::fnal::cd::rms::provider::DPSingleton::Instance(), datagram_client::sl, and string.

                                                         {
     _applicationName = applicationName;
     _partitionNumber = partitionNumber;
 
     char hostname[80];
     memset(hostname, 0, 80);
     if (gethostname(hostname, 79) == 0) {
         _uuidGenerator = new util::UUIDGenerator(hostname);
         std::string tmpString(hostname);
         _myHostName = tmpString;
     }
     else {
         _uuidGenerator = new util::UUIDGenerator("C++DDSConnection");
         _myHostName = "Unknown";
     }
 
     _debugFlag = 0;
     char *debugEnvVar = getenv("RMS_DEBUG");
     if (debugEnvVar != NULL) {
         std::string debugString(debugEnvVar);
         try {
             _debugFlag = boost::lexical_cast<int>(debugString);
         }
         catch (boost::bad_lexical_cast &excpt) {
             std::cerr << "*** ERROR: Invalid RMS_DEBUG value: \""
                       << debugString << "\"." << std::endl;
         }
     }
 
     boost::mutex::scoped_lock sl(_staticMutex);
     ++_instanceCount;
     if (_instanceCount == 1) {
         try {
            DPSingleton::Instance().connect_participant();
         }
         catch (...) {
             std::cerr << "Unable to connect DomainParticipant in "
                       << "DDSConnection constructor." << std::endl;
         }
     }
 }

gov::fnal::cd::rms::provider::DDSConnection::~DDSConnection ( )

Destructor.

Definition at line 73 of file DDSConnection.cpp.

References _instanceCount, _staticMutex, close(), gov::fnal::cd::rms::provider::DPSingleton::disconnect_participant(), gov::fnal::cd::rms::provider::DPSingleton::Instance(), and datagram_client::sl.

                               {
     close();
 
     boost::mutex::scoped_lock sl(_staticMutex);
     --_instanceCount;
     if (_instanceCount == 0) {
         DPSingleton::Instance().disconnect_participant();
     }
 }

Member Function Documentation

void gov::fnal::cd::rms::provider::DDSConnection::close ( )

Closes the connection. After calling this method, the connection should no longer be used since it will no longer be connected to the underlying provider system.

Definition at line 98 of file DDSConnection.cpp.

References _uuidGenerator.

Referenced by getPartitionNumber(), and ~DDSConnection().

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

template<class T >

void gov::fnal::cd::rms::provider::DDSConnection::dumpMessage	(	T const &	message,
		bool	isIncomingMessage
	)

inline

Definition at line 64 of file DDSConnection.h.

References novadaq::convertTimevalToDouble(), om::cout, allTimeWatchdog::endl, and datagram_client::message.

Referenced by receiveMessage(), and sendMessage().

                                                                {
         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;
     }

std::string gov::fnal::cd::rms::provider::DDSConnection::getApplicationName ( ) const

inline

Definition at line 45 of file DDSConnection.h.

References _applicationName.

Referenced by sendMessage().

45 {return _applicationName;}

gov::fnal::cd::rms::provider::DDSConnection::_applicationName

std::string _applicationName

Definition: DDSConnection.h:313

std::string gov::fnal::cd::rms::provider::DDSConnection::getHostName ( )

inline

Definition at line 59 of file DDSConnection.h.

References _myHostName.

                             {
         return _myHostName;
     }

int gov::fnal::cd::rms::provider::DDSConnection::getPartitionNumber ( ) const

inline

Definition at line 46 of file DDSConnection.h.

References _partitionNumber, and close().

Referenced by sendMessage().

46 {return _partitionNumber;}

gov::fnal::cd::rms::provider::DDSConnection::_partitionNumber

int _partitionNumber

Definition: DDSConnection.h:314

const std::string gov::fnal::cd::rms::provider::DDSConnection::getUUID ( )

inline

Get a new human readable UUID from the UUID generator.

Returns: A new UUID.

Definition at line 55 of file DDSConnection.h.

References _uuidGenerator, and gov::fnal::cd::rms::util::UUIDGenerator::getUUIDHex().

Referenced by sendMessage().

                               {
         return _uuidGenerator->getUUIDHex();
     }

template<class T >

void gov::fnal::cd::rms::provider::DDSConnection::receiveMessage	(	unsigned int	receiverHandle,
		T &	message
	)

inline

Receives a message of the specified type from the specified destination.

Parameters

dest	The destination to send the message to.
message	The message received from the destination.

Definition at line 159 of file DDSConnection.h.

References _debugFlag, _readerTable, _registrationMutex, dumpMessage(), makeTrainCVSamples::int, datagram_client::sl, and gov::fnal::cd::rms::provider::DDSTopicReaderClass< TT_ >::take_wait().

                                                                  {
         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);
         }
     }

template<class T >

bool gov::fnal::cd::rms::provider::DDSConnection::receiveMessage	(	unsigned int	receiverHandle,
		T &	message,
		unsigned int	usecTimeout
	)

inline

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.

Parameters

dest	The destination to send the message to.
message	The message received from the destination.
usecTimeout	Amount of time to wait for a message (microseconds).

Returns: true if a message was received, false if not.

Definition at line 190 of file DDSConnection.h.

References _debugFlag, _readerTable, _registrationMutex, om::cout, dumpMessage(), allTimeWatchdog::endl, makeTrainCVSamples::int, datagram_client::message, datagram_client::sl, fabricate::status, and gov::fnal::cd::rms::provider::DDSTopicReaderClass< TT_ >::take_timed_wait().

                                                   {
         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 gov::fnal::cd::rms::provider::DDSConnection::registerReceiver	(	T const &	sampleMessage,
		base::RmsDestination const &	dest
	)

inline

Definition at line 275 of file DDSConnection.h.

References _partitionNumber, _readerCounter, _readerTable, _registrationMutex, gov::fnal::cd::rms::base::RmsDestination::CHANNEL_PROPERTY_NAME, cet::replace_all(), datagram_client::sl, string, and gov::fnal::cd::rms::base::RmsDestination::TARGET_PROPERTY_NAME.

                                                                   {
         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;
     }

template<class T >

unsigned int gov::fnal::cd::rms::provider::DDSConnection::registerSender	(	T const &	sampleMessage,
		base::RmsDestination const &	dest
	)

inline

Definition at line 231 of file DDSConnection.h.

References _partitionNumber, _registrationMutex, _targetChannelTable, _writerCounter, _writerTable, gov::fnal::cd::rms::base::RmsDestination::CHANNEL_PROPERTY_NAME, make_pair(), cet::replace_all(), datagram_client::sl, string, and gov::fnal::cd::rms::base::RmsDestination::TARGET_PROPERTY_NAME.

                                                                 {
         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;
     }

template<class T >

void gov::fnal::cd::rms::provider::DDSConnection::sendMessage	(	unsigned int	senderHandle,
		T &	message
	)

inline

Sends a message of the specified type to the specified destination.

Parameters

dest	The destination to send the message to.
message	The message to be sent to the destination.

Definition at line 105 of file DDSConnection.h.

References _debugFlag, _myHostName, _registrationMutex, _targetChannelTable, _writerTable, dumpMessage(), getApplicationName(), getPartitionNumber(), getUUID(), makeTrainCVSamples::int, datagram_client::sl, string, and gov::fnal::cd::rms::provider::DDSTopicWriterClass< TT_ >::write().

                                                             {
         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);
         }
     }

bool gov::fnal::cd::rms::provider::DDSConnection::supportsDestination ( const base::RmsDestination & dest )

Determine if the given destination can be reached with the DDS provider.

Parameters

dest	The destination to check to see if messages can be sent to it.

Returns: True if the destination can be reached, false otherwise.

Definition at line 89 of file DDSConnection.cpp.

                                                                       {
     return true;
 }

void gov::fnal::cd::rms::provider::DDSConnection::unRegisterReceiver ( unsigned int receiverHandle )

inline

Definition at line 296 of file DDSConnection.h.

References _readerTable, _registrationMutex, makeTrainCVSamples::int, and datagram_client::sl.

                                                          {
         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);
         }
     }

void gov::fnal::cd::rms::provider::DDSConnection::unRegisterSender ( unsigned int senderHandle )

inline

Definition at line 254 of file DDSConnection.h.

References _registrationMutex, _targetChannelTable, _writerTable, makeTrainCVSamples::int, and datagram_client::sl.

                                                      {
         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);
         }
     }

Member Data Documentation

std::string gov::fnal::cd::rms::provider::DDSConnection::_applicationName

private

Definition at line 313 of file DDSConnection.h.

Referenced by DDSConnection(), and getApplicationName().

int gov::fnal::cd::rms::provider::DDSConnection::_debugFlag

private

Definition at line 328 of file DDSConnection.h.

Referenced by DDSConnection(), receiveMessage(), and sendMessage().

int gov::fnal::cd::rms::provider::DDSConnection::_instanceCount = 0

staticprivate

Definition at line 330 of file DDSConnection.h.

Referenced by DDSConnection(), and ~DDSConnection().

std::string gov::fnal::cd::rms::provider::DDSConnection::_myHostName

private

Definition at line 324 of file DDSConnection.h.

Referenced by DDSConnection(), getHostName(), and sendMessage().

int gov::fnal::cd::rms::provider::DDSConnection::_partitionNumber

private

Definition at line 314 of file DDSConnection.h.

Referenced by DDSConnection(), getPartitionNumber(), registerReceiver(), and registerSender().

unsigned int gov::fnal::cd::rms::provider::DDSConnection::_readerCounter

private

Definition at line 322 of file DDSConnection.h.

Referenced by registerReceiver().

std::map<unsigned int, boost::shared_ptr<base::RmsCloseable> > gov::fnal::cd::rms::provider::DDSConnection::_readerTable

private

Definition at line 321 of file DDSConnection.h.

Referenced by receiveMessage(), registerReceiver(), and unRegisterReceiver().

boost::mutex gov::fnal::cd::rms::provider::DDSConnection::_registrationMutex

mutableprivate

Definition at line 326 of file DDSConnection.h.

Referenced by receiveMessage(), registerReceiver(), registerSender(), sendMessage(), unRegisterReceiver(), and unRegisterSender().

boost::mutex gov::fnal::cd::rms::provider::DDSConnection::_staticMutex

staticprivate

Definition at line 331 of file DDSConnection.h.

Referenced by DDSConnection(), and ~DDSConnection().

std::map<unsigned int, std::pair<std::string, std::string> > gov::fnal::cd::rms::provider::DDSConnection::_targetChannelTable

private

Definition at line 318 of file DDSConnection.h.

Referenced by registerSender(), sendMessage(), and unRegisterSender().

util::UUIDGenerator* gov::fnal::cd::rms::provider::DDSConnection::_uuidGenerator

private

Definition at line 312 of file DDSConnection.h.

Referenced by close(), DDSConnection(), and getUUID().

unsigned int gov::fnal::cd::rms::provider::DDSConnection::_writerCounter

private

Definition at line 319 of file DDSConnection.h.

Referenced by registerSender().

std::map<unsigned int, boost::shared_ptr<base::RmsCloseable> > gov::fnal::cd::rms::provider::DDSConnection::_writerTable

private

Definition at line 316 of file DDSConnection.h.

Referenced by registerSender(), sendMessage(), and unRegisterSender().

The documentation for this class was generated from the following files:

/cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N21-01-23/rms/cxx/include/provider/DDSConnection.h
/cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N21-01-23/rms/cxx/src/provider/DDSConnection.cpp

Public Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation